Session Management

Status: Extracted
Purpose: Session Factory pattern, async database operations, and connection pooling.

Session Factory Pattern

Rationale

• Centralizes database engine and session creation
• Enables efficient connection pooling
• Provides single point of configuration for database settings
• Supports dependency injection for testability
• Required foundation for Unit of Work pattern

Implementation

Location: infrastructure/database/session.py

Pattern:

from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
 
def create_session_factory(config: DatabaseConfig) -> async_sessionmaker[AsyncSession]:
    """
    Create an async session factory for database connections.
    
    This function creates an AsyncEngine and returns an async_sessionmaker
    that can be used to create database sessions. The factory manages
    connection pooling and should be created once at application startup.
    """
    engine: AsyncEngine = create_async_engine(
        config.database_url,
        pool_size=config.pool_size,
        max_overflow=config.max_overflow,
        pool_timeout=config.pool_timeout,
        echo=config.echo,
        pool_pre_ping=config.pool_pre_ping,
        future=True,  # Use SQLAlchemy 2.0 style
    )
    
    return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

Usage

Application Startup:

• Session factory created at application startup
• Stored in app.state.session_factory for dependency access
• Uses SQLAlchemy async_sessionmaker with AsyncEngine
• All database operations use sessions from the factory

Session Creation:

# Get session factory from app state
session_factory = app.state.session_factory
 
# Create a session
async with session_factory() as session:
    # Use session for database operations
    result = await session.execute(query)

Async Database Operations

Rationale

• Aligns with FastAPI's async model for better performance
• Enables non-blocking I/O for database operations
• Improves concurrency and scalability
• Better resource utilization for I/O-bound operations
• Industry standard for async Python web applications

Implementation

Driver: Uses asyncpg driver instead of psycopg2-binary

Session Type: All database operations use AsyncSession from SQLAlchemy

Pattern:

• Repository interfaces and implementations are async
• Command handlers are async to support async repository calls
• Unit of Work manages async sessions
• All database queries use await for async execution

Example:

async def get_user_by_id(self, user_id: str) -> User | None:
    """Get user by ID."""
    async with self.session_factory() as session:
        result = await session.execute(
            select(UserModel).where(UserModel.id == user_id)
        )
        user_model = result.scalar_one_or_none()
        return self._to_domain(user_model) if user_model else None

Connection Pooling

Configuration

Pool Settings:

• pool_size: Number of connections to maintain in pool (default: 5)
• max_overflow: Maximum overflow connections (default: 10)
• pool_timeout: Timeout for getting connection from pool (default: 30)
• pool_pre_ping: Test connections before using (default: True)

Benefits:

• Reuses connections for better performance
• Limits concurrent connections to database
• Handles connection failures gracefully
• Reduces connection overhead

Session Lifecycle

Session Creation

1. Factory Creation: Session factory created at application startup
2. Session Request: Session created from factory when needed
3. Transaction Start: Transaction begins when session is used
4. Operations: Database operations performed within session
5. Commit/Rollback: Transaction committed or rolled back
6. Session Close: Session closed and returned to pool

Best Practices

• Use Context Managers: Always use async with for session management
• Single Transaction: Keep all related operations in one session
• Error Handling: Ensure sessions are closed even on errors
• Connection Pooling: Let factory manage connection pooling

Related Documents

• Migration principles (all stacks) — Tool-agnostic migration rules
• Migration patterns — Alembic — Python / SQLAlchemy workflow
• Migration patterns — Drizzle Kit — TypeScript workflow
• Unit of Work Pattern - Transaction management
• Repository Implementation - Repository patterns

Reference implementation: infrastructure/database/session.py in any FastAPI service.