Akarshan Gandotra

Posted on Jul 14 • Edited on Jul 15

🔌 Building Resilient Database Operations with Async SQLAlchemy + CircuitBreaker

#aiobreaker #sqlalchemy #fastapi #python

When building production applications, database failures are inevitable. Network issues, high load, or database maintenance can cause your application to hang or crash. The Circuit Breaker pattern provides an elegant solution to handle these failures gracefully while maintaining system stability. 🛡️

⚡ What is the Circuit Breaker Pattern?

The Circuit Breaker pattern prevents your application from repeatedly attempting operations that are likely to fail. Just like an electrical circuit breaker, it monitors failures and "trips" when a threshold is reached, preventing further calls until the system recovers. 🔄

🚦 Circuit Breaker States

🟢 Closed: Normal operation, requests flow through
🔴 Open: Failures exceeded threshold, requests fail immediately
🟡 Half-Open: Testing if the system has recovered

🗄️ Why Use Circuit Breakers for Database Operations?

Database operations are particularly vulnerable to:

⏰ Network timeouts
🏊 Connection pool exhaustion
📈 Database server overload
🔧 Temporary unavailability during maintenance

Without circuit breakers, your application might:

💥 Exhaust connection pools
🌊 Create cascading failures
😤 Provide poor user experience with long timeouts
💸 Waste resources on doomed operations

🏗️ Implementation: Database Circuit Breaker in Python

Let's build a robust database layer with circuit breaker protection using SQLAlchemy and asyncio.

🔧 Core Circuit Breaker Component

from aiobreaker import CircuitBreaker, CircuitBreakerError
from datetime import timedelta
import logging

class DatabaseCircuitBreakerError(Exception):
    """Custom exception for circuit breaker failures."""
    def __init__(self, message: str = None):
        self.message = message or "Database service is temporarily unavailable"
        self.error_code = "DB_CIRCUIT_BREAKER_OPEN"
        super().__init__(self.message)

class DatabaseCircuitBreaker:
    """Circuit breaker for database operations."""

    def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 60):
        self.circuit_breaker = CircuitBreaker(
            fail_max=failure_threshold,
            timeout_duration=timedelta(seconds=recovery_timeout),
            exclude=self._get_excluded_exceptions(),
        )

    def _get_excluded_exceptions(self) -> tuple:
        """Exclude application errors, include infrastructure errors."""
        return (
            IntegrityError,        # Data validation errors
            InvalidRequestError,   # SQL syntax errors
            ValueError,           # Application logic errors
            TypeError,            # Programming errors
        )

    async def call(self, func, *args, **kwargs):
        try:
            return await self.circuit_breaker.call_async(func, *args, **kwargs)
        except CircuitBreakerError as e:
            logger.warning("Circuit breaker is open, cannot execute database operation")
            raise DatabaseCircuitBreakerError() from e

🔧 Understanding Circuit Breaker Configuration

Let's break down the key parameters of the CircuitBreaker initialization:

CircuitBreaker(
    fail_max=failure_threshold,
    timeout_duration=timedelta(seconds=recovery_timeout),
    exclude=self._get_excluded_exceptions(),
)

fail_max - The failure threshold that triggers the circuit breaker:

After this many consecutive failures, the circuit breaker "trips" and opens
Example: With fail_max=5, the circuit breaker opens after 5 database connection failures

timeout_duration - How long the circuit breaker stays open:

When open, it waits this duration before testing recovery (half-open state)
Example: With timeout_duration=60, it waits 60 seconds before retrying

exclude - Exceptions that should NOT count toward failures:

These are "application errors" rather than infrastructure failures
Infrastructure failures (network timeouts, connection refused) should trip the breaker
Application failures (bad SQL, invalid data) should not trip the breaker

🚦 Circuit Breaker State Flow

🟢 CLOSED (Normal operation)
    ↓ (5 consecutive failures)
🔴 OPEN (Blocking all requests)
    ↓ (60 seconds timeout)
🟡 HALF-OPEN (Testing recovery)
    ↓ (Success) → 🟢 CLOSED
    ↓ (Failure) → 🔴 OPEN

🎯 Session Manager with Circuit Breaker

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from contextlib import asynccontextmanager
import asyncio

class SessionManager:
    """Database session manager with circuit breaker protection."""

    def __init__(self):
        self.engine = None
        self.session_factory = None
        self.circuit_breaker = DatabaseCircuitBreaker()
        self._is_initialized = False
        self._engine_lock = asyncio.Lock()

    async def init_db(self):
        """Initialize database engine and session factory."""
        async with self._engine_lock:
            if self._is_initialized:
                return

            # Database URL construction
            database_url = self._build_database_url()

            # Create engine with optimized settings
            self.engine = create_async_engine(
                database_url,
                pool_size=10,
                max_overflow=20,
                pool_pre_ping=True,
                pool_recycle=3600,
                pool_timeout=30,
                connect_args={
                    "server_settings": {
                        "statement_timeout": "5000"  # 5 second timeout
                    }
                }
            )

            self.session_factory = async_sessionmaker(
                bind=self.engine,
                expire_on_commit=False,
                autoflush=False,
                class_=AsyncSession,
            )

            await self._health_check()
            self._is_initialized = True
            logger.info("Database initialized successfully")

    async def _health_check(self):
        """Verify database connectivity."""
        try:
            async with self.engine.begin() as conn:
                await conn.execute(text("SELECT 1"))
            logger.info("Database health check passed")
        except Exception as e:
            logger.error(f"Database health check failed: {e}")
            raise RuntimeError(f"Database health check failed: {e}") from e

    @asynccontextmanager
    async def get_session(self):
        """Get a circuit breaker protected session."""
        session = await self.circuit_breaker.call(self._create_session)
        protected_session = CircuitBreakerSession(session, self.circuit_breaker)

        try:
            yield protected_session
        except Exception as e:
            await session.rollback()
            logger.exception(f"Exception in DB session, rolling back: {e}")
            raise
        finally:
            await session.close()

    async def _create_session(self) -> AsyncSession:
        if not self._is_initialized:
            await self.init_db()

        if not self.session_factory:
            raise RuntimeError("Database session factory is not initialized.")

        return self.session_factory()

🛡️ Protected Session Wrapper

The CircuitBreakerSession wraps the original SQLAlchemy session and applies circuit breaker protection to all database operations. This ensures that every database call goes through the circuit breaker, providing consistent protection across your application.

🔍 Key Design Principles

Selective Protection: Only async operations that can fail due to infrastructure issues are protected:

execute() - Running SQL queries
commit() - Persisting transactions
rollback() - Reverting transactions
flush() - Synchronizing with database
refresh() - Reloading object state

Synchronous Operations: Methods like add() and delete() are not protected because they only modify in-memory state and don't communicate with the database until a flush/commit occurs.

🏗️ Implementation Details

class CircuitBreakerSession:
    """Session wrapper that applies circuit breaker to all operations."""

    def __init__(self, session: AsyncSession, circuit_breaker: DatabaseCircuitBreaker):
        self._session = session
        self._circuit_breaker = circuit_breaker
        self._is_closed = False

    def _ensure_not_closed(self):
        """Prevent operations on closed sessions."""
        if self._is_closed:
            raise RuntimeError("Session is already closed")

    async def execute(self, statement, **kwargs):
        """Execute statement with circuit breaker protection."""
        self._ensure_not_closed()

        async def _execute():
            return await self._session.execute(statement, **kwargs)

        return await self._circuit_breaker.call(_execute)

    async def commit(self):
        """Commit transaction with circuit breaker protection."""
        self._ensure_not_closed()
        return await self._circuit_breaker.call(self._session.commit)

    async def rollback(self):
        """Rollback transaction with circuit breaker protection."""
        self._ensure_not_closed()
        return await self._circuit_breaker.call(self._session.rollback)

    # Synchronous operations don't need circuit breaker
    def add(self, instance):
        """Add object to session (in-memory only)."""
        self._ensure_not_closed()
        return self._session.add(instance)

    def delete(self, instance):
        """Mark object for deletion (in-memory only)."""
        self._ensure_not_closed()
        return self._session.delete(instance)

🔄 How It Works

Wrapping Pattern: The session wrapper intercepts all method calls
State Checking: Each method verifies the session isn't closed
Selective Protection: Only async database operations get circuit breaker protection
Transparent Interface: Behaves exactly like a regular SQLAlchemy session
Error Propagation: Circuit breaker errors are raised when the breaker is open

🚀 Usage with FastAPI

from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()
sessionmanager = SessionManager()

async def get_db():
    """FastAPI dependency for database sessions."""
    async with sessionmanager.get_session() as session:
        yield session

@app.get("/users/{user_id}")
async def get_user(user_id: int, db = Depends(get_db)):
    try:
        result = await db.execute(
            select(User).where(User.id == user_id)
        )
        user = result.scalar_one_or_none()

        if not user:
            raise HTTPException(status_code=404, detail="User not found")

        return user

    except DatabaseCircuitBreakerError:
        raise HTTPException(
            status_code=503, 
            detail="Database service temporarily unavailable"
        )

@app.on_event("startup")
async def startup():
    await sessionmanager.init_db()

@app.on_event("shutdown")
async def shutdown():
    await sessionmanager.close()

⚙️ Configuration Best Practices

🌍 Environment Variables

# Circuit breaker settings
DB_CIRCUIT_BREAKER_FAILURE_THRESHOLD = 5  # Trip after 5 failures
DB_CIRCUIT_BREAKER_RECOVERY_TIMEOUT = 60   # Wait 60 seconds before retry

# Connection pool settings
POOL_SIZE = 10
MAX_OVERFLOW = 20
POOL_RECYCLE = 3600
POOL_TIMEOUT = 30

⚖️ Benefits and Trade-offs

✅ Benefits

🛡️ Improved resilience: Prevents cascading failures
😊 Better user experience: Fast failure instead of timeouts
🔐 Resource protection: Prevents connection pool exhaustion
🔄 Automatic recovery: Tests system health periodically

⚠️ Trade-offs

🔧 Added complexity: More moving parts to monitor
❌ False positives: May block valid requests during recovery
⚙️ Configuration overhead: Requires tuning for your workload

✅ Demo Video

🎯 Conclusion

The Circuit Breaker pattern is essential for building resilient database layers in production applications. By implementing circuit breakers around your database operations, you can:

🚫 Prevent cascading failures
💪 Improve system stability
🎯 Provide better error handling
🛡️ Protect system resources

The implementation shown here provides a solid foundation that you can adapt to your specific needs. Remember to monitor circuit breaker behavior and tune the configuration based on your application's requirements. 📊

What patterns do you use for database resilience in your applications? Share your experiences in the comments below! 💬

DEV Community