Async Service Layer with __aenter__ / __aexit__ Session Lifecycle

What? (Concept Overview)

The service layer owns the database transaction boundary: it creates the AsyncSession, hands it to a repository, commits the whole unit of work on success, rolls back on exception, and closes the session deterministically. By implementing the async context-manager protocol (__aenter__ / __aexit__), every service call site becomes a clean async with CustomerService() as svc: block, mirroring the lifecycle of a unit-of-work.

Project Context

The FCA Support Agent has BaseService at app/services/base.py and concrete subclasses (CustomerService, AccountService, TransactionService, ConversationService, MessageService, ProductService, FAQService, RAGService, SecurityService, CacheService). The bulk seed script (app/seed_database.py) relies on this pattern to commit-and-close between customer batches; committing inside one big transaction would defeat idempotency when a single customer row fails.

How? (Quick Reference Blocks)

3.1 The Async Context-Manager Base Class

# app/services/base.py
from contextlib import asynccontextmanager
from app.database import AsyncSessionLocal
 
class BaseService:
    def __init__(self) -> None:
        self.session = None
 
    async def __aenter__(self):
        # Open a fresh AsyncSession from the global factory on entry.
        self.session = AsyncSessionLocal()
        # Hand the session to a repository configured for this service.
        return self
 
    async def __aexit__(self, exc_type, exc, tb) -> None:
        try:
            if exc_type is None:
                # No exception → commit pending work.
                await self.session.commit()
            else:
                # Exception → roll back ALL pending changes.
                await self.session.rollback()
        finally:
            # Always close the session regardless of commit/rollback.
            await self.session.close()
            self.session = None
 
    async def commit(self) -> None:
        await self.session.commit()
 
    async def rollback(self) -> None:
        await self.session.rollback()

3.2 Concrete Service & Repository Pairing

# app/services/customer.py (illustrative)
from app.repositories.customer import CustomerRepository
from app.services.base import BaseService
 
class CustomerService(BaseService):
    @property
    def repo(self) -> CustomerRepository:
        # Built lazily so the session is open by the time it's called.
        return CustomerRepository(self.session)
 
    async def create_customer(self, **data) -> "Customer":
        return await self.repo.create(data)
 
    async def find_by_email(self, email: str) -> "Customer | None":
        return await self.repo.find_by_email(email)

3.3 Canonical Call Site

# app/seed_database.py — bulk seed (slot-by-slot commit)
async with CustomerService() as service:
    customer = await service.create_customer(**data)
    await service.commit()    # explicit checkpoint for batching

Why? (Parameter Breakdown)

• __aenter__ opens the session, __aexit__ closes it — Without the context manager, every call site would need to remember commit/rollback/close in the right order. The protocol makes session lifecycle a language-level guarantee.
• exc_type is None branch is the commit signal — Python sets exc_type=None on a clean exit. By branching on it inside __aexit__, the service commits only when the async with block completed without error — eliminating the “half-committed transaction on exception” footgun.
• finally: await self.session.close() — Guarantees the connection is returned to the pool even if commit/rollback raises. This is the single most important line; without it Postgres can lock up under bursty traffic.
• Lazy repo property, not eager init in __init__ — The session is None until __aenter__. If you stash the repo in __init__, the repo holds a stale (closed) session and every call AttributeErrors. Lazy resolution keeps the repo bound to a live session.
• Explicit commit()/rollback() methods on the service — Lets callers checkpoint inside a long-running service call (e.g., 1000-row bulk insert) without giving up the lifecycle safety of __aexit__.
• One service per unit-of-work, not per request handler — The pattern composes: many services can be entered sequentially in a single request (each committing its own UoW), and a service is trivially mockable by replacing AsyncSessionLocal in tests.

Common Pitfalls

1. Stashing the repo in __init__ instead of as a property. Result: bad session reference on first use. Use @property to bind to the open session on demand.
2. Catching exceptions inside the async with block and not re-raising. This silently swallows business errors AND bypasses the rollback path. If you must catch, re-raise. Or use a narrower try inside, not around the whole block.

Real-World Interview Prep

Q1: How would you make this pattern transactionally compose across multiple services (e.g., create a customer AND an account in one atomic operation)?

A: Two options. Option A — share a session: pass an open AsyncSession into both services’ constructors (override __init__(session) and skip the context manager), wrap both calls in async with UnitOfWork(): parent that owns the session. Option B — outbox pattern: have each service write a row to an event_outbox table; a background worker drains the outbox atomically. Option A is simpler; Option B is mandatory when the two services live in different processes (microservices). Choose A when services share a process.

Q2: How do you test a service that opens its own session?

A: Override the global AsyncSessionLocal factory in a pytest fixture:

@pytest.fixture
async def session_factory():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    return async_sessionmaker(engine, expire_on_commit=False)
 
@pytest.fixture
async def customer_service(monkeypatch, session_factory):
    monkeypatch.setattr("app.services.base.AsyncSessionLocal", session_factory)
    async with CustomerService() as svc:
        yield svc

The monkeypatch.setattr is mandatory — AsyncSessionLocal is referenced at import time inside BaseService.__aenter__.

Q3: Why split service and repository instead of letting services open sessions per call?

A: Three reasons. (1) Testability — repository tests don’t need a service, service tests don’t need SQL. (2) Composability — you can stack two repos in one service without nesting transactions. (3) Single-transaction semantics — a service is “one unit-of-work”; merging repos into services produces a god-class per resource that mixes persistence with business logic, and impossible-to-reason-about error paths. The discipline mirrors the Hexagonal/Ports-and-Adapters model where services are “use cases” and repos are “ports”.

Top-to-Bottom Code Walkthrough (app/services/base.py + concrete services)

The “async service layer” pattern fixes the biggest bug in web apps: forgetting to commit, or commit then forget to close. It uses __aenter__/__aexit__ to make a session a contextual resource with automatic lifecycle.

BaseService (app/services/base.py)

class BaseService:
    def __init__(self):
        self.session: AsyncSession | None = None
        self.committed = False
 
    async def __aenter__(self):
        self.session = AsyncSessionLocal()
        return self
 
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        try:
            if exc_type is not None:
                await self.session.rollback()
                self.committed = False
            elif not self.committed:
                await self.session.rollback()  # safety: no commit == rollback
        finally:
            await self.session.close()
 
    async def commit(self) -> None:
        await self.session.commit()
        self.committed = True
 
    async def rollback(self) -> None:
        await self.session.rollback()
        self.committed = False

Key design points:

1. No-commit-equals-rollback in __aexit__: if you forgot to call commit() the work is rolled back. Safer than leaving a transaction open.
2. Exception-safe: any exception inside the async with block triggers rollback() before close.
3. self.committed flag lets us tell apart a successful explicit commit from an unintentional no-op.

Concrete example: CustomerService

class CustomerService(BaseService):
    async def create_customer(self, **data):
        repo = CustomerRepository(self.session, Customer)
        hsh = security_service.get_password_hash(data.pop("password"))
        data["hashed_password"] = hsh
        return await repo.create(data)

The service owns the session because: (a) it owns the transaction boundary, (b) it can do multiple repo calls in one transaction (e.g., create customer + log audit event), (c) rollback is at the service level, not per repo.

Multi-step example: AccountService.create_with_initial_deposit

class AccountService(BaseService):
    async def create_with_initial_deposit(self, account_data, deposit_amount):
        account_repo = AccountRepository(self.session, Account)
        txn_repo = TransactionRepository(self.session, Transaction)
 
        account = await account_repo.create(account_data)
        await txn_repo.create({
            "account_id": account.id,
            "amount": deposit_amount,
            "description": "Opening deposit",
        })
        # Single commit for both — atomic.
        await self.commit()

If txn_repo.create fails, the account INSERT is rolled back. No orphan accounts. The unit-of-work owns the entire operation.

Usage pattern in route handlers

@router.post("/customers")
async def create_customer(
    body: CustomerBody,
    db: AsyncSession = Depends(get_db),  # FALLBACK — not used
):
    async with CustomerService() as svc:
        customer = await svc.create_customer(**body.dict())
        await svc.commit()
        return customer.id

get_db (the FastAPI dependency) is technically available but services don’t use it — they own the session. Cleaner ownership.

Common Pitfalls

Forgetting to call commit() — the __aexit__ rollout rolls back. If you want persistence, commit before exit.

Catching the exception inside the async with and not re-raising — __aexit__ won’t see the exception, won’t rollback. Always re-raise.

Calling await self.session.close() directly bypasses the safety net. Use async with exclusively.

Real-World Interview Prep

Q1: Why __aenter__/__aexit__ instead of a with body that creates and closes the session itself?

A: Inheritance. CustomerService(BaseService) doesn’t need to redefine session lifecycle — just override methods. The base class enforces the contract.

Q2: How do you test a service that owns its session?

A: Subclass BaseService, override __aenter__ to yield a session bound to a Postgres test container. Replace AsyncSessionLocal in the test conftest. The service code is unchanged.

Q3: What happens if __aexit__’s rollback itself raises?

A: A “rollback failed” exception shadows the original exception — very confusing. Wrap the rollback: try: await self.session.rollback() except Exception: log_and_continue. Always close, never re-raise during teardown.