Python Database Patterns Best Practices¶

# python/01-connection-pooling.py

"""
Database connection pooling patterns and management
"""

from typing import List, Dict, Any, Optional, Union, Tuple
from dataclasses import dataclass, asdict
from enum import Enum
import asyncio
import time
import threading
from contextlib import asynccontextmanager, contextmanager
from datetime import datetime, timedelta
import logging
from queue import Queue, Empty
import weakref
import psycopg2
from psycopg2 import pool
from sqlalchemy import create_engine, text
from sqlalchemy.pool import QueuePool, StaticPool
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
import asyncpg
import aioredis
from redis import ConnectionPool as RedisConnectionPool

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ConnectionState(Enum):
    """Connection state enumeration"""
    IDLE = "idle"
    BUSY = "busy"
    CLOSED = "closed"
    ERROR = "error"

@dataclass
class ConnectionMetrics:
    """Connection metrics"""
    total_connections: int = 0
    active_connections: int = 0
    idle_connections: int = 0
    failed_connections: int = 0
    connection_requests: int = 0
    connection_wait_time: float = 0.0
    last_activity: datetime = None

class DatabaseConnectionPool:
    """Advanced database connection pool"""

    def __init__(self, connection_string: str, min_connections: int = 5, 
                 max_connections: int = 20, connection_timeout: int = 30):
        self.connection_string = connection_string
        self.min_connections = min_connections
        self.max_connections = max_connections
        self.connection_timeout = connection_timeout
        self.pool = None
        self.metrics = ConnectionMetrics()
        self.connection_lock = threading.Lock()
        self.connections = {}
        self.initialize_pool()

    def initialize_pool(self) -> None:
        """Initialize connection pool"""
        try:
            self.pool = psycopg2.pool.ThreadedConnectionPool(
                minconn=self.min_connections,
                maxconn=self.max_connections,
                dsn=self.connection_string,
                connect_timeout=self.connection_timeout
            )
            logger.info(f"Connection pool initialized with {self.min_connections}-{self.max_connections} connections")
        except Exception as e:
            logger.error(f"Failed to initialize connection pool: {e}")
            raise

    def get_connection(self) -> psycopg2.extensions.connection:
        """Get connection from pool"""
        start_time = time.time()

        try:
            connection = self.pool.getconn()
            if connection:
                self.metrics.active_connections += 1
                self.metrics.connection_requests += 1
                self.metrics.connection_wait_time += time.time() - start_time
                self.metrics.last_activity = datetime.utcnow()

                # Store connection reference
                with self.connection_lock:
                    self.connections[id(connection)] = {
                        'connection': connection,
                        'state': ConnectionState.BUSY,
                        'created_at': datetime.utcnow()
                    }

                logger.debug(f"Connection acquired: {id(connection)}")
                return connection
            else:
                raise Exception("Failed to get connection from pool")

        except Exception as e:
            self.metrics.failed_connections += 1
            logger.error(f"Failed to get connection: {e}")
            raise

    def return_connection(self, connection: psycopg2.extensions.connection) -> None:
        """Return connection to pool"""
        try:
            connection_id = id(connection)

            with self.connection_lock:
                if connection_id in self.connections:
                    self.connections[connection_id]['state'] = ConnectionState.IDLE
                    del self.connections[connection_id]

            self.pool.putconn(connection)
            self.metrics.active_connections -= 1
            self.metrics.idle_connections += 1

            logger.debug(f"Connection returned: {connection_id}")

        except Exception as e:
            logger.error(f"Failed to return connection: {e}")

    def close_connection(self, connection: psycopg2.extensions.connection) -> None:
        """Close connection"""
        try:
            connection_id = id(connection)

            with self.connection_lock:
                if connection_id in self.connections:
                    self.connections[connection_id]['state'] = ConnectionState.CLOSED
                    del self.connections[connection_id]

            connection.close()
            self.metrics.active_connections -= 1

            logger.debug(f"Connection closed: {connection_id}")

        except Exception as e:
            logger.error(f"Failed to close connection: {e}")

    def get_pool_status(self) -> Dict[str, Any]:
        """Get pool status and metrics"""
        with self.connection_lock:
            active_connections = len([conn for conn in self.connections.values() 
                                       if conn['state'] == ConnectionState.BUSY])
            idle_connections = len([conn for conn in self.connections.values() 
                                  if conn['state'] == ConnectionState.IDLE])

        return {
            "min_connections": self.min_connections,
            "max_connections": self.max_connections,
            "active_connections": active_connections,
            "idle_connections": idle_connections,
            "total_connections": len(self.connections),
            "failed_connections": self.metrics.failed_connections,
            "connection_requests": self.metrics.connection_requests,
            "avg_wait_time": self.metrics.connection_wait_time / max(self.metrics.connection_requests, 1),
            "last_activity": self.metrics.last_activity.isoformat() if self.metrics.last_activity else None
        }

    def close_pool(self) -> None:
        """Close all connections in pool"""
        try:
            if self.pool:
                self.pool.closeall()
                logger.info("Connection pool closed")
        except Exception as e:
            logger.error(f"Failed to close connection pool: {e}")

class AsyncDatabasePool:
    """Async database connection pool"""

    def __init__(self, connection_string: str, min_connections: int = 5,
                 max_connections: int = 20):
        self.connection_string = connection_string
        self.min_connections = min_connections
        self.max_connections = max_connections
        self.pool = None
        self.metrics = ConnectionMetrics()
        self.connection_lock = asyncio.Lock()
        self.connections = {}

    async def initialize_pool(self) -> None:
        """Initialize async connection pool"""
        try:
            self.pool = await asyncpg.create_pool(
                self.connection_string,
                min_size=self.min_connections,
                max_size=self.max_connections,
                command_timeout=60
            )
            logger.info(f"Async connection pool initialized with {self.min_connections}-{self.max_connections} connections")
        except Exception as e:
            logger.error(f"Failed to initialize async connection pool: {e}")
            raise

    async def get_connection(self) -> asyncpg.Connection:
        """Get async connection from pool"""
        start_time = time.time()

        try:
            connection = await self.pool.acquire()
            self.metrics.active_connections += 1
            self.metrics.connection_requests += 1
            self.metrics.connection_wait_time += time.time() - start_time
            self.metrics.last_activity = datetime.utcnow()

            logger.debug(f"Async connection acquired: {id(connection)}")
            return connection

        except Exception as e:
            self.metrics.failed_connections += 1
            logger.error(f"Failed to get async connection: {e}")
            raise

    async def return_connection(self, connection: asyncpg.Connection) -> None:
        """Return async connection to pool"""
        try:
            await self.pool.release(connection)
            self.metrics.active_connections -= 1
            self.metrics.idle_connections += 1

            logger.debug(f"Async connection returned: {id(connection)}")

        except Exception as e:
            logger.error(f"Failed to return async connection: {e}")

    async def close_pool(self) -> None:
        """Close async connection pool"""
        try:
            if self.pool:
                await self.pool.close()
                logger.info("Async connection pool closed")
        except Exception as e:
            logger.error(f"Failed to close async connection pool: {e}")

class SQLAlchemyConnectionManager:
    """SQLAlchemy connection management"""

    def __init__(self, connection_string: str, pool_size: int = 10,
                 max_overflow: int = 20, pool_timeout: int = 30):
        self.connection_string = connection_string
        self.engine = None
        self.session_factory = None
        self.pool_size = pool_size
        self.max_overflow = max_overflow
        self.pool_timeout = pool_timeout
        self.initialize_engine()

    def initialize_engine(self) -> None:
        """Initialize SQLAlchemy engine with connection pooling"""
        try:
            self.engine = create_engine(
                self.connection_string,
                poolclass=QueuePool,
                pool_size=self.pool_size,
                max_overflow=self.max_overflow,
                pool_timeout=self.pool_timeout,
                pool_pre_ping=True,
                pool_recycle=3600,  # Recycle connections every hour
                echo=False
            )

            self.session_factory = sessionmaker(bind=self.engine)
            logger.info(f"SQLAlchemy engine initialized with pool size {self.pool_size}")

        except Exception as e:
            logger.error(f"Failed to initialize SQLAlchemy engine: {e}")
            raise

    @contextmanager
    def get_session(self):
        """Get database session with automatic cleanup"""
        session = self.session_factory()
        try:
            yield session
            session.commit()
        except Exception as e:
            session.rollback()
            logger.error(f"Database session error: {e}")
            raise
        finally:
            session.close()

    def get_engine_status(self) -> Dict[str, Any]:
        """Get engine and pool status"""
        pool = self.engine.pool
        return {
            "pool_size": pool.size(),
            "checked_in": pool.checkedin(),
            "checked_out": pool.checkedout(),
            "overflow": pool.overflow(),
            "invalid": pool.invalid()
        }

class AsyncSQLAlchemyManager:
    """Async SQLAlchemy connection management"""

    def __init__(self, connection_string: str, pool_size: int = 10,
                 max_overflow: int = 20):
        self.connection_string = connection_string
        self.engine = None
        self.session_factory = None
        self.pool_size = pool_size
        self.max_overflow = max_overflow
        self.initialize_engine()

    def initialize_engine(self) -> None:
        """Initialize async SQLAlchemy engine"""
        try:
            self.engine = create_async_engine(
                self.connection_string,
                pool_size=self.pool_size,
                max_overflow=self.max_overflow,
                pool_pre_ping=True,
                pool_recycle=3600,
                echo=False
            )

            self.session_factory = sessionmaker(
                bind=self.engine,
                class_=AsyncSession,
                expire_on_commit=False
            )
            logger.info(f"Async SQLAlchemy engine initialized with pool size {self.pool_size}")

        except Exception as e:
            logger.error(f"Failed to initialize async SQLAlchemy engine: {e}")
            raise

    @asynccontextmanager
    async def get_session(self):
        """Get async database session with automatic cleanup"""
        session = self.session_factory()
        try:
            yield session
            await session.commit()
        except Exception as e:
            await session.rollback()
            logger.error(f"Async database session error: {e}")
            raise
        finally:
            await session.close()

class RedisConnectionManager:
    """Redis connection management"""

    def __init__(self, host: str = "localhost", port: int = 6379,
                 db: int = 0, max_connections: int = 20):
        self.host = host
        self.port = port
        self.db = db
        self.max_connections = max_connections
        self.pool = None
        self.initialize_pool()

    def initialize_pool(self) -> None:
        """Initialize Redis connection pool"""
        try:
            self.pool = RedisConnectionPool(
                host=self.host,
                port=self.port,
                db=self.db,
                max_connections=self.max_connections,
                retry_on_timeout=True
            )
            logger.info(f"Redis connection pool initialized with {self.max_connections} connections")
        except Exception as e:
            logger.error(f"Failed to initialize Redis connection pool: {e}")
            raise

    def get_connection(self):
        """Get Redis connection from pool"""
        try:
            from redis import Redis
            return Redis(connection_pool=self.pool)
        except Exception as e:
            logger.error(f"Failed to get Redis connection: {e}")
            raise

    def get_pool_status(self) -> Dict[str, Any]:
        """Get Redis pool status"""
        return {
            "max_connections": self.max_connections,
            "created_connections": self.pool.created_connections,
            "available_connections": self.pool.available_connections,
            "in_use_connections": self.pool.in_use_connections
        }

# Usage examples
def example_connection_pooling():
    """Example connection pooling usage"""
    # PostgreSQL connection pool
    pg_pool = DatabaseConnectionPool(
        connection_string="postgresql://user:pass@localhost/db",
        min_connections=5,
        max_connections=20
    )

    # Get connection
    connection = pg_pool.get_connection()
    try:
        cursor = connection.cursor()
        cursor.execute("SELECT 1")
        result = cursor.fetchone()
        print(f"Query result: {result}")
    finally:
        pg_pool.return_connection(connection)

    # Get pool status
    status = pg_pool.get_pool_status()
    print(f"Pool status: {status}")

    # SQLAlchemy connection manager
    sqlalchemy_manager = SQLAlchemyConnectionManager(
        connection_string="postgresql://user:pass@localhost/db",
        pool_size=10,
        max_overflow=20
    )

    # Use session
    with sqlalchemy_manager.get_session() as session:
        result = session.execute(text("SELECT 1")).fetchone()
        print(f"SQLAlchemy result: {result}")

    # Redis connection manager
    redis_manager = RedisConnectionManager(
        host="localhost",
        port=6379,
        max_connections=20
    )

    # Use Redis connection
    redis_client = redis_manager.get_connection()
    redis_client.set("test_key", "test_value")
    value = redis_client.get("test_key")
    print(f"Redis value: {value}")

    # Close pools
    pg_pool.close_pool()

# python/02-transaction-management.py

"""
Database transaction management patterns and ACID compliance
"""

from typing import List, Dict, Any, Optional, Union, Tuple, Callable
from dataclasses import dataclass, asdict
from enum import Enum
import asyncio
import time
from contextlib import asynccontextmanager, contextmanager
from datetime import datetime, timedelta
import logging
from functools import wraps
import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_READ_COMMITTED, ISOLATION_LEVEL_REPEATABLE_READ
from sqlalchemy.orm import Session
from sqlalchemy.exc import SQLAlchemyError
import asyncpg
from asyncpg import Connection

logger = logging.getLogger(__name__)

class TransactionIsolation(Enum):
    """Transaction isolation levels"""
    READ_UNCOMMITTED = "READ UNCOMMITTED"
    READ_COMMITTED = "READ COMMITTED"
    REPEATABLE_READ = "REPEATABLE READ"
    SERIALIZABLE = "SERIALIZABLE"

class TransactionManager:
    """Database transaction manager"""

    def __init__(self, connection_pool: DatabaseConnectionPool):
        self.connection_pool = connection_pool
        self.transaction_metrics = {
            "total_transactions": 0,
            "successful_transactions": 0,
            "failed_transactions": 0,
            "rollback_count": 0,
            "avg_transaction_time": 0.0
        }

    @contextmanager
    def transaction(self, isolation_level: TransactionIsolation = TransactionIsolation.READ_COMMITTED):
        """Context manager for database transactions"""
        connection = None
        start_time = time.time()

        try:
            # Get connection from pool
            connection = self.connection_pool.get_connection()

            # Set isolation level
            if isolation_level == TransactionIsolation.READ_COMMITTED:
                connection.set_isolation_level(ISOLATION_LEVEL_READ_COMMITTED)
            elif isolation_level == TransactionIsolation.REPEATABLE_READ:
                connection.set_isolation_level(ISOLATION_LEVEL_REPEATABLE_READ)

            # Start transaction
            connection.autocommit = False

            self.transaction_metrics["total_transactions"] += 1

            yield connection

            # Commit transaction
            connection.commit()
            self.transaction_metrics["successful_transactions"] += 1

            transaction_time = time.time() - start_time
            self._update_avg_transaction_time(transaction_time)

            logger.debug(f"Transaction committed in {transaction_time:.3f}s")

        except Exception as e:
            # Rollback transaction
            if connection:
                connection.rollback()
                self.transaction_metrics["rollback_count"] += 1
                logger.error(f"Transaction rolled back: {e}")

            self.transaction_metrics["failed_transactions"] += 1
            raise

        finally:
            # Return connection to pool
            if connection:
                self.connection_pool.return_connection(connection)

    def _update_avg_transaction_time(self, transaction_time: float) -> None:
        """Update average transaction time"""
        total_transactions = self.transaction_metrics["total_transactions"]
        current_avg = self.transaction_metrics["avg_transaction_time"]

        # Calculate new average
        new_avg = ((current_avg * (total_transactions - 1)) + transaction_time) / total_transactions
        self.transaction_metrics["avg_transaction_time"] = new_avg

    def get_transaction_metrics(self) -> Dict[str, Any]:
        """Get transaction metrics"""
        total = self.transaction_metrics["total_transactions"]
        success_rate = (self.transaction_metrics["successful_transactions"] / total * 100) if total > 0 else 0

        return {
            **self.transaction_metrics,
            "success_rate": success_rate
        }

class AsyncTransactionManager:
    """Async database transaction manager"""

    def __init__(self, connection_pool: AsyncDatabasePool):
        self.connection_pool = connection_pool
        self.transaction_metrics = {
            "total_transactions": 0,
            "successful_transactions": 0,
            "failed_transactions": 0,
            "rollback_count": 0,
            "avg_transaction_time": 0.0
        }

    @asynccontextmanager
    async def transaction(self, isolation_level: str = "READ COMMITTED"):
        """Async context manager for database transactions"""
        connection = None
        start_time = time.time()

        try:
            # Get connection from pool
            connection = await self.connection_pool.get_connection()

            # Start transaction
            async with connection.transaction(isolation=isolation_level):
                self.transaction_metrics["total_transactions"] += 1

                yield connection

                # Transaction will be committed automatically
                self.transaction_metrics["successful_transactions"] += 1

                transaction_time = time.time() - start_time
                self._update_avg_transaction_time(transaction_time)

                logger.debug(f"Async transaction committed in {transaction_time:.3f}s")

        except Exception as e:
            # Transaction will be rolled back automatically
            self.transaction_metrics["failed_transactions"] += 1
            self.transaction_metrics["rollback_count"] += 1
            logger.error(f"Async transaction rolled back: {e}")
            raise

        finally:
            # Return connection to pool
            if connection:
                await self.connection_pool.return_connection(connection)

    def _update_avg_transaction_time(self, transaction_time: float) -> None:
        """Update average transaction time"""
        total_transactions = self.transaction_metrics["total_transactions"]
        current_avg = self.transaction_metrics["avg_transaction_time"]

        new_avg = ((current_avg * (total_transactions - 1)) + transaction_time) / total_transactions
        self.transaction_metrics["avg_transaction_time"] = new_avg

    def get_transaction_metrics(self) -> Dict[str, Any]:
        """Get transaction metrics"""
        total = self.transaction_metrics["total_transactions"]
        success_rate = (self.transaction_metrics["successful_transactions"] / total * 100) if total > 0 else 0

        return {
            **self.transaction_metrics,
            "success_rate": success_rate
        }

class SQLAlchemyTransactionManager:
    """SQLAlchemy transaction manager"""

    def __init__(self, session_factory):
        self.session_factory = session_factory
        self.transaction_metrics = {
            "total_transactions": 0,
            "successful_transactions": 0,
            "failed_transactions": 0,
            "rollback_count": 0,
            "avg_transaction_time": 0.0
        }

    @contextmanager
    def transaction(self):
        """SQLAlchemy transaction context manager"""
        session = self.session_factory()
        start_time = time.time()

        try:
            self.transaction_metrics["total_transactions"] += 1

            yield session

            # Commit transaction
            session.commit()
            self.transaction_metrics["successful_transactions"] += 1

            transaction_time = time.time() - start_time
            self._update_avg_transaction_time(transaction_time)

            logger.debug(f"SQLAlchemy transaction committed in {transaction_time:.3f}s")

        except SQLAlchemyError as e:
            # Rollback transaction
            session.rollback()
            self.transaction_metrics["rollback_count"] += 1
            self.transaction_metrics["failed_transactions"] += 1
            logger.error(f"SQLAlchemy transaction rolled back: {e}")
            raise

        finally:
            session.close()

    def _update_avg_transaction_time(self, transaction_time: float) -> None:
        """Update average transaction time"""
        total_transactions = self.transaction_metrics["total_transactions"]
        current_avg = self.transaction_metrics["avg_transaction_time"]

        new_avg = ((current_avg * (total_transactions - 1)) + transaction_time) / total_transactions
        self.transaction_metrics["avg_transaction_time"] = new_avg

    def get_transaction_metrics(self) -> Dict[str, Any]:
        """Get transaction metrics"""
        total = self.transaction_metrics["total_transactions"]
        success_rate = (self.transaction_metrics["successful_transactions"] / total * 100) if total > 0 else 0

        return {
            **self.transaction_metrics,
            "success_rate": success_rate
        }

def transactional(isolation_level: TransactionIsolation = TransactionIsolation.READ_COMMITTED):
    """Decorator for transactional methods"""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            # This would be implemented with actual transaction manager
            # For now, just execute the function
            return func(*args, **kwargs)
        return wrapper
    return decorator

class DatabaseRepository:
    """Repository pattern for database operations"""

    def __init__(self, transaction_manager: TransactionManager):
        self.transaction_manager = transaction_manager

    def create_user(self, user_data: Dict[str, Any]) -> int:
        """Create user with transaction"""
        with self.transaction_manager.transaction() as conn:
            cursor = conn.cursor()

            # Insert user
            cursor.execute("""
                INSERT INTO users (name, email, created_at) 
                VALUES (%s, %s, %s) RETURNING id
            """, (user_data['name'], user_data['email'], datetime.utcnow()))

            user_id = cursor.fetchone()[0]

            # Insert user profile
            cursor.execute("""
                INSERT INTO user_profiles (user_id, bio, avatar_url) 
                VALUES (%s, %s, %s)
            """, (user_id, user_data.get('bio', ''), user_data.get('avatar_url', '')))

            return user_id

    def get_user(self, user_id: int) -> Optional[Dict[str, Any]]:
        """Get user by ID"""
        with self.transaction_manager.transaction() as conn:
            cursor = conn.cursor()

            cursor.execute("""
                SELECT u.id, u.name, u.email, u.created_at, 
                       p.bio, p.avatar_url
                FROM users u
                LEFT JOIN user_profiles p ON u.id = p.user_id
                WHERE u.id = %s
            """, (user_id,))

            result = cursor.fetchone()
            if result:
                return {
                    'id': result[0],
                    'name': result[1],
                    'email': result[2],
                    'created_at': result[3],
                    'bio': result[4],
                    'avatar_url': result[5]
                }
            return None

    def update_user(self, user_id: int, user_data: Dict[str, Any]) -> bool:
        """Update user with transaction"""
        with self.transaction_manager.transaction() as conn:
            cursor = conn.cursor()

            # Update user
            cursor.execute("""
                UPDATE users 
                SET name = %s, email = %s, updated_at = %s
                WHERE id = %s
            """, (user_data['name'], user_data['email'], datetime.utcnow(), user_id))

            # Update user profile
            cursor.execute("""
                UPDATE user_profiles 
                SET bio = %s, avatar_url = %s
                WHERE user_id = %s
            """, (user_data.get('bio', ''), user_data.get('avatar_url', ''), user_id))

            return cursor.rowcount > 0

    def delete_user(self, user_id: int) -> bool:
        """Delete user with transaction"""
        with self.transaction_manager.transaction() as conn:
            cursor = conn.cursor()

            # Delete user profile first (foreign key constraint)
            cursor.execute("DELETE FROM user_profiles WHERE user_id = %s", (user_id,))

            # Delete user
            cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))

            return cursor.rowcount > 0

class DatabaseMigrationManager:
    """Database migration management"""

    def __init__(self, connection_pool: DatabaseConnectionPool):
        self.connection_pool = connection_pool
        self.migrations = []
        self.initialize_migrations_table()

    def initialize_migrations_table(self) -> None:
        """Initialize migrations tracking table"""
        with self.connection_pool.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS schema_migrations (
                    id SERIAL PRIMARY KEY,
                    version VARCHAR(255) UNIQUE NOT NULL,
                    applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)
            conn.commit()

    def add_migration(self, version: str, sql: str) -> None:
        """Add migration"""
        self.migrations.append({
            'version': version,
            'sql': sql,
            'applied': False
        })

    def run_migrations(self) -> List[str]:
        """Run pending migrations"""
        applied_migrations = []

        with self.connection_pool.get_connection() as conn:
            cursor = conn.cursor()

            # Get applied migrations
            cursor.execute("SELECT version FROM schema_migrations")
            applied_versions = {row[0] for row in cursor.fetchall()}

            # Run pending migrations
            for migration in self.migrations:
                if migration['version'] not in applied_versions:
                    try:
                        cursor.execute(migration['sql'])
                        cursor.execute(
                            "INSERT INTO schema_migrations (version) VALUES (%s)",
                            (migration['version'],)
                        )
                        conn.commit()
                        applied_migrations.append(migration['version'])
                        logger.info(f"Applied migration: {migration['version']}")

                    except Exception as e:
                        conn.rollback()
                        logger.error(f"Failed to apply migration {migration['version']}: {e}")
                        raise

        return applied_migrations

# Usage examples
def example_transaction_management():
    """Example transaction management usage"""
    # Create connection pool
    connection_pool = DatabaseConnectionPool("postgresql://user:pass@localhost/db")

    # Create transaction manager
    transaction_manager = TransactionManager(connection_pool)

    # Use transaction
    with transaction_manager.transaction() as conn:
        cursor = conn.cursor()
        cursor.execute("INSERT INTO users (name, email) VALUES (%s, %s)", ("John", "john@example.com"))
        cursor.execute("INSERT INTO user_profiles (user_id, bio) VALUES (%s, %s)", (1, "Software Developer"))

    # Get transaction metrics
    metrics = transaction_manager.get_transaction_metrics()
    print(f"Transaction metrics: {metrics}")

    # Repository pattern
    repository = DatabaseRepository(transaction_manager)

    # Create user
    user_data = {
        'name': 'Jane Doe',
        'email': 'jane@example.com',
        'bio': 'Data Scientist',
        'avatar_url': 'https://example.com/avatar.jpg'
    }
    user_id = repository.create_user(user_data)
    print(f"Created user with ID: {user_id}")

    # Get user
    user = repository.get_user(user_id)
    print(f"Retrieved user: {user}")

    # Database migrations
    migration_manager = DatabaseMigrationManager(connection_pool)

    # Add migration
    migration_manager.add_migration("001_create_users_table", """
        CREATE TABLE users (
            id SERIAL PRIMARY KEY,
            name VARCHAR(255) NOT NULL,
            email VARCHAR(255) UNIQUE NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # Run migrations
    applied_migrations = migration_manager.run_migrations()
    print(f"Applied migrations: {applied_migrations}")

    # Close connection pool
    connection_pool.close_pool()

# 1. Connection pooling
pg_pool = DatabaseConnectionPool("postgresql://user:pass@localhost/db")
connection = pg_pool.get_connection()
pg_pool.return_connection(connection)

# 2. Transaction management
transaction_manager = TransactionManager(pg_pool)
with transaction_manager.transaction() as conn:
    cursor = conn.cursor()
    cursor.execute("INSERT INTO users (name) VALUES (%s)", ("John",))

# 3. SQLAlchemy integration
sqlalchemy_manager = SQLAlchemyConnectionManager("postgresql://user:pass@localhost/db")
with sqlalchemy_manager.get_session() as session:
    result = session.execute(text("SELECT 1")).fetchone()

# 4. Repository pattern
repository = DatabaseRepository(transaction_manager)
user_id = repository.create_user({"name": "John", "email": "john@example.com"})

# 5. Database migrations
migration_manager = DatabaseMigrationManager(pg_pool)
migration_manager.add_migration("001_create_table", "CREATE TABLE users (id SERIAL PRIMARY KEY)")
applied = migration_manager.run_migrations()

# Complete database setup
def setup_database_patterns():
    """Setup complete database patterns environment"""

    # Connection pooling
    pg_pool = DatabaseConnectionPool("postgresql://user:pass@localhost/db")
    async_pool = AsyncDatabasePool("postgresql://user:pass@localhost/db")

    # Transaction management
    transaction_manager = TransactionManager(pg_pool)
    async_transaction_manager = AsyncTransactionManager(async_pool)

    # SQLAlchemy integration
    sqlalchemy_manager = SQLAlchemyConnectionManager("postgresql://user:pass@localhost/db")
    async_sqlalchemy_manager = AsyncSQLAlchemyManager("postgresql://user:pass@localhost/db")

    # Redis connection
    redis_manager = RedisConnectionManager("localhost", 6379)

    # Repository pattern
    repository = DatabaseRepository(transaction_manager)

    # Migration management
    migration_manager = DatabaseMigrationManager(pg_pool)

    print("Database patterns setup complete!")