__aenter__ / __aexit__ Session LifecycleSERVICE LAYER ASYNC SESSION — Data Flow
async with Service() as svc:
│
▼
svc.session ── AsyncSession bound to the unit-of-work
│
▼
svc.repo.method() ──► repo.session.execute(sql)
│
▼
commit() ── once at __aexit__
│
▼
rollback() ── on any exception
│
▼
close() ── session released back to the poolThe 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.
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.
# 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()# 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)# 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__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.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.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__.AsyncSessionLocal in tests.__init__ instead of as a property. Result: bad session reference on first use. Use @property to bind to the open session on demand.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.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.
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 svcThe monkeypatch.setattr is mandatory — AsyncSessionLocal is referenced at import time inside BaseService.__aenter__.
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”.
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 = FalseKey design points:
__aexit__: if you forgot to call commit() the work is rolled back. Safer than leaving a transaction open.async with block triggers rollback() before close.self.committed flag lets us tell apart a successful explicit commit from an unintentional no-op.CustomerServiceclass 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.
AccountService.create_with_initial_depositclass 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.
@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.idget_db (the FastAPI dependency) is technically available but services don’t use it — they own the session. Cleaner ownership.
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.
__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.
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.
__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.
__aexit__ runs await self.session.commit() but the FastAPI handler already returned.await svc.commit() at end of handler; remove auto-commit.commit only on explicit await svc.commit().except Exception: await svc.rollback() in route handler.AsyncSession; each’s aexit commits.