remove SQLAlchemy ORM (#380)

* refactor: remove SQLAlchemy ORM for direct asyncpg driver

- Switch from SQLAlchemy to asyncpg for raw PostgreSQL access
- Add connection pooling with min=10 max=POOL_SIZE
- Set 300s max idle timeout for connections
- Implement proper pool lifecycle via FastAPI lifespan

* refactor: migrate from SQLAlchemy to asyncpg for database operations

- Replace SQLAlchemy ORM with direct asyncpg queries for better performance
- Add DeviceRequest Pydantic model for request validation
- Implement proper error handling for database operations
- Add FastAPI lifespan context manager

Testing:
- Validated all PostgreSQL operations with neon.tech database
- Confirmed working: inserts, error handling, metrics collection

TODO:
- Need to test memcached operations

* refactor: replace SQLAlchemy models with raw SQL schema

- Remove models.py as we no longer use SQLAlchemy ORM
- Add schema.sql for direct PostgreSQL table definitions

* remove SQLAlchemy dependencies

* Add async support set operations in aiomcache
- Updated requirements.txt

Author: techUdayMungalpara

lessons/231/python-app/db.py

@@ -1,27 +1,83 @@
+import logging
 import os
-from typing import Annotated
+from contextlib import asynccontextmanager
+from typing import Annotated, AsyncGenerator
+
+import asyncpg
+from fastapi import Depends, FastAPI
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
 
-from fastapi import Depends
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 
 POSTGRES_URI = os.environ["POSTGRES_URI"]
 POSTGRES_POOL_SIZE = int(os.environ["POSTGRES_POOL_SIZE"])
 
 
-engine = create_async_engine(
-    POSTGRES_URI,
-    echo=False,
-    pool_pre_ping=False,
-    pool_size=POSTGRES_POOL_SIZE,
-    max_overflow=0,
-)
+class Database:
+    def __init__(self):
+        self.pool = None
+
+    async def create_pool(self):
+        """Create connection pool if it doesn't exist"""
+        if self.pool is None:
+            try:
+                self.pool = await asyncpg.create_pool(
+                    POSTGRES_URI,
+                    min_size=10,
+                    max_size=POSTGRES_POOL_SIZE,
+                    max_inactive_connection_lifetime=300,
+                )
+                logger.info(f"Database pool created: {self.pool}")
+            except asyncpg.exceptions.PostgresError as e:
+                logging.error(f"Error creating PostgreSQL connection pool: {e}")
+                raise ValueError("Failed to create PostgreSQL connection pool")
+            except Exception as e:
+                logging.error(f"Unexpected error while creating connection pool: {e}")
+                raise
+
+    @asynccontextmanager
+    async def get_connection(self) -> AsyncGenerator[asyncpg.Connection, None]:
+        """Get database connection from pool"""
+        if not self.pool:
+            await self.create_pool()
+        async with self.pool.acquire() as connection:
+            logger.info("Connection acquired from pool")
+            yield connection
+            logger.info("Connection released back to pool")
+
+    async def close(self):
+        """Close the pool when shutting down"""
+        if self.pool:
+            await self.pool.close()
+            logger.info("Database pool closed")
+            self.pool = None
+
+
+db = Database()
+
 
-async_session = async_sessionmaker(engine, expire_on_commit=False)
+async def get_db() -> AsyncGenerator[asyncpg.Connection, None]:
+    async with db.get_connection() as conn:
+        yield conn
 
 
-async def get_postgres_session():
-    async with async_session() as session:
-        yield session
+PostgresDep = Annotated[asyncpg.Connection, Depends(get_db)]
 
 
-PostgresDep = Annotated[AsyncSession, Depends(get_postgres_session)]
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Lifespan context manager for database connection"""
+    print(" Starting up database connection...")
+    try:
+        await db.create_pool()
+        logger.info(" Database pool created successfully")
+        yield
+    except Exception as e:
+        logger.info(f"Failed to create database pool: {e}")
+        raise
+    finally:
+        # Shutdown: close all connections
+        logger.info(" Shutting down database connection...")
+        await db.close()
+        logger.info(" Database connections closed")

lessons/231/python-app/main.py

@@ -4,20 +4,22 @@
 import time
 import uuid
 
-from fastapi import FastAPI
+import aiomcache
+import orjson
+from asyncpg import PostgresError
+from fastapi import FastAPI, HTTPException
 from fastapi.responses import ORJSONResponse, PlainTextResponse
 from prometheus_client import make_asgi_app
+from pydantic import BaseModel
 from pymemcache.client.base import Client
-from sqlalchemy import insert
-from asyncer import asyncify
-from db import PostgresDep
+
+from db import PostgresDep, lifespan
 from metrics import H
-from models import Device
 
-app = FastAPI()
+app = FastAPI(lifespan=lifespan)
 
 MEMCACHED_HOST = os.environ["MEMCACHED_HOST"]
-cache_client = Client(MEMCACHED_HOST)
+cache_client = aiomcache.Client(MEMCACHED_HOST)
 
 metrics_app = make_asgi_app()
 app.mount("/metrics", metrics_app)
@@ -75,34 +77,56 @@ def get_devices():
     return devices
 
 
+class DeviceRequest(BaseModel):
+    mac: str
+    firmware: str
+
+
 @app.post("/api/devices", status_code=201, response_class=ORJSONResponse)
-async def create_device(device: Device, session: PostgresDep) -> Device:
-    # To match Go implementation instead of using SQLAlchemy  factory.
-    now = datetime.datetime.now(datetime.timezone.utc)
-    device_uuid = uuid.uuid4()
-
-    stmt = (
-        insert(Device)
-        .values(
-            uuid=device_uuid,
-            mac=device.mac,
-            firmware=device.firmware,
-            created_at=now,
-            updated_at=now,
+async def create_device(device: DeviceRequest, conn: PostgresDep):
+    try:
+        now = datetime.datetime.now(datetime.timezone.utc)
+        device_uuid = uuid.uuid4()
+
+        insert_query = """
+            INSERT INTO python_device (uuid, mac, firmware, created_at, updated_at)
+            VALUES ($1, $2, $3, $4, $5)
+            RETURNING id, uuid, mac, firmware, created_at, updated_at;
+            """
+
+        start_time = time.perf_counter()
+
+        row = await conn.fetchrow(
+            insert_query, device_uuid, device.mac, device.firmware, now, now
+        )
+
+        H.labels(op="insert", db="postgres").observe(time.perf_counter() - start_time)
+
+        device_dict = dict(row)
+
+        # Measure cache operation
+        start_time = time.perf_counter()
+
+        await cache_client.set(
+            device_uuid.bytes,
+            orjson.dumps(device_dict),
+            exptime=20,
+        )
+
+        H.labels(op="set", db="memcache").observe(time.perf_counter() - start_time)
+
+        return device_dict
+
+    except PostgresError as e:
+        raise HTTPException(
+            status_code=500, detail="Database error occurred while creating device"
+        )
+    except aiomcache.exceptions.ClientException as e:
+        raise HTTPException(
+            status_code=500,
+            detail="Memcached Database error occurred while creating device",
+        )
+    except Exception as e:
+        raise HTTPException(
+            status_code=500, detail="An unexpected error occurred while creating device"
         )
-        .returning(Device)
-    )
-
-    # Measure the same insert operation as in Go
-    start_time = time.perf_counter()
-    device_result = await session.execute(stmt)
-    device_dict = device_result.mappings().one()
-    await session.commit()
-    H.labels(op="insert", db="postgres").observe(time.perf_counter() - start_time)
-
-    # Measure the same set operation as in Go
-    start_time = time.perf_counter()
-    cache_client.set(str(device_uuid), dict(device_dict), expire=20)
-    H.labels(op="set", db="memcache").observe(time.perf_counter() - start_time)
-
-    return device_dict

lessons/231/python-app/models.py

@@ -1,3 +1,4 @@
+aiomcache==0.8.2
 annotated-types==0.7.0
 anyio==4.7.0
 asyncpg==0.30.0
@@ -7,7 +8,6 @@ dnspython==2.7.0
 email_validator==2.2.0
 fastapi==0.115.6
 fastapi-cli==0.0.6
-greenlet==3.1.1
 h11==0.14.0
 httpcore==1.0.7
 httptools==0.6.4
@@ -34,12 +34,11 @@ rich==13.9.4
 rich-toolkit==0.12.0
 shellingham==1.5.4
 sniffio==1.3.1
-SQLAlchemy==2.0.36
 starlette==0.41.3
 typer==0.15.1
 typing_extensions==4.12.2
 uvicorn==0.32.1
 uvloop==0.21.0
 watchfiles==1.0.0
 websockets==14.1
-asyncer==0.0.8
+asyncer==0.0.8

lessons/231/python-app/requirements.txt

@@ -0,0 +1,14 @@
+CREATE TABLE IF NOT EXISTS python_device (
+    id SERIAL PRIMARY KEY,
+    uuid UUID DEFAULT NULL,
+    mac VARCHAR(255) DEFAULT NULL,
+    firmware VARCHAR(255) DEFAULT NULL,
+    created_at TIMESTAMP
+    WITH
+        TIME ZONE DEFAULT CURRENT_TIMESTAMP,
+        updated_at TIMESTAMP
+    WITH
+        TIME ZONE DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX IF NOT EXISTS idx_device_uuid ON python_device (uuid);