Convert database I/O to AsyncSession

In this lesson you will convert database I/O to AsyncSession. Work with the Equipment Rental API in the downloadable lab. The reference is a complete solution with separate lesson checks, so you can inspect the answer, make a deliberate local change and verify its behavior.

AsyncSession changes how database I/O is awaited, while the business invariant remains the same. reserve_async still locks equipment, reads active quantity, rejects insufficient capacity and inserts one rental within one transaction. execute, scalar, get and flush perform I/O and are awaited. Adding await at the boundary should preserve the response shape and error behavior of the synchronous service.

Create an asynchronous engine with an async-capable driver and an async_sessionmaker. Each request opens and closes its own AsyncSession. expire_on_commit=False helps the reference construct and return an already-loaded response without unexpected post-commit attribute I/O. Explicit eager loading is still necessary for relationships; a serializer must not accidentally try implicit async database work. Transaction completion precedes the success response just as it did synchronously. Convert one boundary at a time and compare behavior through the same contracts. An async rewrite is not permission to relax capacity checks, hide failures or share mutable session state across concurrently running tasks.

Worked source: equipment/services.py, reserve_async.

Compare reserve and reserve_async side by side. Run lesson13: the async service reserves one unit, rejects a subsequent request for two, and a separate synchronous session confirms only one active unit was committed. This cross-check verifies database visibility as well as the response model.

async def reserve_async(session, equipment_id, owner_id, quantity):
    if quantity < 1:
        raise DomainError("invalid_quantity", "Request at least one unit.", 422)
    async with session.begin():
        item = await session.scalar(
            select(Equipment).where(Equipment.id == equipment_id).with_for_update()
        )
        if item is None:
            raise missing()
        used = await session.scalar(
            select(func.coalesce(func.sum(Rental.quantity), 0)).where(
                Rental.equipment_id == equipment_id, Rental.status == "active"
            )
        )
        if used + quantity > item.quantity:
            raise DomainError(
                "capacity_unavailable", "Not enough equipment units are available."
            )
        rental = Rental(equipment_id=equipment_id, owner_id=owner_id, quantity=quantity)
        session.add(rental)
        await session.flush()
        result = RentalOut.model_validate(rental)
    return result

python run_checks.py -k lesson13