Address code review feedback: use lifespan, improve exception handling

Co-authored-by: jason810496 <[email protected]>

Author: Copilot

examples/fastapi_pub_sub/api.py

@@ -6,10 +6,10 @@
 - Creating orders and sending them to a message queue
 """
 from typing import Generator
-from contextlib import contextmanager
+from contextlib import contextmanager, asynccontextmanager
 
 from fastapi import FastAPI, Depends, HTTPException
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime
 from sqlalchemy.orm import Session, sessionmaker, declarative_base
 from datetime import datetime
@@ -47,6 +47,8 @@ class OrderCreate(BaseModel):
 
 
 class OrderResponse(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    
     id: int
     customer_name: str
     product_name: str
@@ -55,40 +57,41 @@ class OrderResponse(BaseModel):
     created_at: datetime
     message_id: int
 
-    class Config:
-        from_attributes = True
-
-
-# FastAPI app
-app = FastAPI(title="Order Management with PGMQ")
-
-
-# Database dependency
-def get_db() -> Generator[Session, None, None]:
-    """Database session dependency."""
-    db = SessionLocal()
-    try:
-        yield db
-    finally:
-        db.close()
-
 
-@app.on_event("startup")
-def startup_event():
+# Lifespan context manager for startup/shutdown
+@asynccontextmanager
+async def lifespan(app: FastAPI):
     """Initialize database tables and PGMQ queue on startup."""
-    # Create tables if they don't exist
+    # Startup
     Base.metadata.create_all(bind=engine)
 
     # Initialize PGMQ queue
     with SessionLocal() as session:
         op.check_pgmq_ext(session=session, commit=True)
 
-        # Create queue if it doesn't exist (will not fail if exists)
+        # Create queue if it doesn't exist
         try:
             op.create_queue(QUEUE_NAME, session=session, commit=True)
         except Exception:
             # Queue might already exist, which is fine
             pass
+    
+    yield
+    
+    # Shutdown (if needed)
+
+
+# FastAPI app with lifespan
+app = FastAPI(title="Order Management with PGMQ", lifespan=lifespan)
+
+# Database dependency
+def get_db() -> Generator[Session, None, None]:
+    """Database session dependency."""
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
 
 
 @app.post("/orders", response_model=OrderResponse, status_code=201)

examples/fastapi_pub_sub/consumer.py

@@ -115,7 +115,11 @@ async def main():
     async_engine = create_async_engine(DATABASE_URL)
     async_session_maker = sessionmaker(bind=async_engine, class_=AsyncSession)
 
-    # Create PGMQueue instance manually to avoid event loop issues
+    # Note: Manual PGMQueue setup to avoid event loop conflicts
+    # PGMQueue.__init__ tries to run a nested event loop which conflicts
+    # with asyncio.run(). This is a known limitation when using PGMQueue
+    # in an async context manager like asyncio.run().
+    # For proper usage, consider using PGMQOperation methods directly with sessions.
     pgmq = PGMQueue.__new__(PGMQueue)
     pgmq.engine = async_engine
     pgmq.session_maker = async_session_maker

examples_tests/test_fastapi_pub_sub.py

@@ -48,8 +48,10 @@ def setup_api_app(sync_database_url, test_queue_name):
         api.op.check_pgmq_ext(session=session, commit=True)
         try:
             api.op.create_queue(test_queue_name, session=session, commit=True)
-        except Exception:
-            pass
+        except Exception as e:
+            # Queue already exists from a previous test run
+            import logging
+            logging.warning(f"Could not create queue (may already exist): {e}")
 
     yield api
 
@@ -58,8 +60,9 @@ def setup_api_app(sync_database_url, test_queue_name):
         # Drop the test queue
         try:
             api.op.drop_queue(test_queue_name, session=session, commit=True)
-        except Exception:
-            pass
+        except Exception as e:
+            import logging
+            logging.warning(f"Could not drop queue: {e}")
 
         # Drop tables
         session.execute(text("DROP TABLE IF EXISTS orders CASCADE"))
@@ -169,8 +172,10 @@ async def test_consumer_processing(async_database_url, sync_database_url, test_q
         op.check_pgmq_ext(session=session, commit=True)
         try:
             op.create_queue(test_queue_name, session=session, commit=True)
-        except Exception:
-            pass  # Queue might already exist
+        except Exception as e:
+            # Queue already exists from a previous test run
+            import logging
+            logging.warning(f"Could not create queue (may already exist): {e}")
 
     test_message = {
         "order_id": 12345,