Backend Live Coding

# ======================================================================
# Q: When do you write async def and when def in FastAPI?
#
# FastAPI runs def in a threadpool, so blocking code (sync SQLAlchemy, requests)
# is safe there. async def runs on the event loop: if you put a blocking call
# inside, the whole server freezes. So async def only when the whole chain is
# async (AsyncSession, httpx.AsyncClient, redis.asyncio).
# ======================================================================

# WRONG: blocks the event loop for all users
@app.get('/bad')
async def bad():
    return requests.get(URL, timeout=10).json()

# RIGHT 1: sync endpoint -> threadpool
@app.get('/ok-sync')
def ok_sync():
    return requests.get(URL, timeout=10).json()

# RIGHT 2: fully async
@app.get('/ok-async')
async def ok_async(client: httpx.AsyncClient = Depends(get_http_client)):
    r = await client.get(URL, timeout=10)
    return r.json()

# ======================================================================
# Q: Write a FastAPI endpoint with validation, a dependency for the DB session
# and correct status codes.
#
# Pydantic does the validation and generates OpenAPI docs on its own.
# response_model filters the fields so internal fields do not leak. The
# tenant_id check is inside the query (404, not 403). Depends corresponds to
# DRF's authentication/permission classes.
# ======================================================================

from fastapi import FastAPI, Depends, HTTPException, status
from pydantic import BaseModel, Field, ConfigDict

app = FastAPI()

class AssetIn(BaseModel):
    hostname: str = Field(min_length=1, max_length=255)
    ip: str
    criticality: int = Field(ge=1, le=5)

class AssetOut(AssetIn):
    model_config = ConfigDict(from_attributes=True)    # from an ORM object
    id: int

@app.post('/v1/assets', response_model=AssetOut,
          status_code=status.HTTP_201_CREATED)
def create_asset(data: AssetIn,
                 session: Session = Depends(get_session),
                 user: User = Depends(require_permission('assets:write'))):
    asset = Asset(**data.model_dump(), tenant_id=user.tenant_id)
    session.add(asset)
    session.flush()                     # so it gets an id before the commit
    return asset

@app.get('/v1/assets/{asset_id}', response_model=AssetOut)
def get_asset(asset_id: int, session: Session = Depends(get_session),
              user: User = Depends(current_user)):
    asset = session.scalar(select(Asset).where(
        Asset.id == asset_id, Asset.tenant_id == user.tenant_id))
    if asset is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND)
    return asset

# ======================================================================
# Q: How do you work with Alembic? What do you watch for in autogenerate?
#
# Autogenerate is a suggestion, not a final file. It does not always detect a
# rename (it sees it as drop + add, so you lose data), changes to server
# defaults, or some constraints. I read every migration before committing. For
# large tables I do the same as in Django: expand/contract, index CONCURRENTLY,
# backfill in batches.
# In CI I run alembic upgrade head on an empty database and check that
# autogenerate produces no new changes (the model and the database agree). I do
# the same in Django with makemigrations --check.
# ======================================================================

def upgrade():
    # CONCURRENTLY cannot run inside a transaction
    with op.get_context().autocommit_block():
        op.create_index('ix_events_created_at', 'events', ['created_at'],
                        postgresql_concurrently=True)

def downgrade():
    with op.get_context().autocommit_block():
        op.drop_index('ix_events_created_at', table_name='events',
                      postgresql_concurrently=True)

# ======================================================================
# Q: How do you do an atomic stock decrement and locking in SQLAlchemy?
#
# The same three solutions I know from Django (chapter 4). The difference is
# that optimistic locking is built into SQLAlchemy with version_id_col.
# ======================================================================

from sqlalchemy import update

# 1) Atomic update in the database (like F() in Django)
res = session.execute(
    update(Product)
    .where(Product.id == pid, Product.stock >= qty)
    .values(stock=Product.stock - qty))
if res.rowcount == 0:
    raise OutOfStock(pid)

# 2) Pessimistic lock (like select_for_update)
product = session.scalars(
    select(Product).where(Product.id == pid).with_for_update()).one()

# 3) Optimistic locking: built in
class Product(Base):
    version: Mapped[int] = mapped_column(default=1)
    __mapper_args__ = {'version_id_col': version}   # StaleDataError on conflict

# ======================================================================
# Q: How do you manage the Session in a web application?
#
# One Session per request (or per Celery task), never global and never shared
# between threads. Commit at the end if everything went well, rollback on
# exception, always close. expire_on_commit=False helps when you return objects
# after the commit (e.g. in the response).
# ======================================================================

engine = create_engine(settings.DATABASE_URL, pool_size=10, max_overflow=20,
                       pool_pre_ping=True)        # checks for dead connections
SessionLocal = sessionmaker(engine, expire_on_commit=False)

def get_session():                               # FastAPI dependency
    with SessionLocal() as session:
        with session.begin():                     # commit or rollback automatically
            yield session