Backend Live Coding

# ======================================================================
# Q: Write the upsert so the import can be re-run without duplicate records.
#
# Idempotency rests on a natural key (external_id) with a unique constraint. If
# the job dies midway, I re-run it from the start without harm. The where clause
# prevents an old file from overwriting newer data.
# ======================================================================

from sqlalchemy.dialects.postgresql import insert as pg_insert

def upsert_events(session, rows):
    stmt = pg_insert(Event).values(rows)
    stmt = stmt.on_conflict_do_update(
        index_elements=[Event.external_id],         # unique constraint
        set_={'severity': stmt.excluded.severity,
              'host': stmt.excluded.host,
              'updated_at': func.now()},
        where=Event.occurred_at <= stmt.excluded.occurred_at)  # newer data only
    session.execute(stmt)

# ======================================================================
# Q: You have to import a 5 GB CSV with 30 million rows. How do you do it
# without running out of memory?
#
# Streaming with generators: I read line by line, validate and write in batches.
# Never readlines() or loading the whole file with pandas.read_csv. For a clean
# import the fastest option is PostgreSQL COPY into a staging table and then
# INSERT ... SELECT with an upsert.
# ======================================================================

import csv
from itertools import batched           # Python 3.12+
from pydantic import BaseModel, ValidationError

class EventRow(BaseModel):
    external_id: str
    host: str
    severity: int
    occurred_at: datetime

def parse(path):
    with open(path, newline='', encoding='utf-8') as f:
        for lineno, raw in enumerate(csv.DictReader(f), start=2):
            try:
                yield EventRow(**raw).model_dump(), None
            except ValidationError as e:
                yield None, {'line': lineno, 'raw': raw, 'errors': e.errors()}

def import_file(session, path, import_id):
    for chunk in batched(parse(path), 5000):
        good = [row for row, err in chunk if row]
        bad = [err for row, err in chunk if err]
        if good:
            upsert_events(session, good)
        if bad:
            session.execute(insert(RejectedRow),
                            [{'import_id': import_id, **b} for b in bad])
        session.commit()                 # small transactions, not one huge one

# ======================================================================
# Q: Design RBAC. How do you check permissions on each endpoint?
#
# Tables users, roles, permissions, user_roles, role_permissions. Permissions
# use the resource:action format (assets:read, reports:export). The check is
# done on permissions, not on role names, so a role can change without a deploy.
# Default deny.
# RBAC says "what is allowed". You also need an ownership/tenant check (does the
# object belong to the user?), otherwise you have BOLA, number 1 on the OWASP
# API Top 10. When the rules depend on attributes (department, country,
# classification level), we move to ABAC or a policy engine (e.g. OPA).
# ======================================================================

def require_permission(perm: str):
    def checker(user: User = Depends(current_user)) -> User:
        if perm not in user.permissions:      # a set, loaded once per request
            raise HTTPException(status.HTTP_403_FORBIDDEN)
        return user
    return checker

@app.get('/v1/reports/{rid}/export')
def export_report(rid: int,
                  user: User = Depends(require_permission('reports:export')),
                  session: Session = Depends(get_session)):
    report = get_scoped(session, Report, rid, user.tenant_id)   # + tenant check
    audit_log(session, user, 'reports:export', report.id)       # compliance
    return stream_csv(report)

# ======================================================================
# Q: Write JWT issuing and verification. What do you watch out for?
#
# (1) A fixed list of algorithms, because the classic attack is alg=none or
# switching RS256 to HS256. (2) Check exp, aud, iss. (3) A short lifetime for
# the access token (5-15 minutes) and a refresh token with rotation: each
# refresh issues a new one, and if an old one is used again, I invalidate the
# whole family (a sign of theft). (4) I put no sensitive data in the JWT: it is
# base64, not encrypted. (5) For logout before expiry: a denylist of the jti in
# Redis with a TTL equal to the remaining lifetime.
# ======================================================================

import jwt, uuid
from datetime import datetime, timedelta, timezone

ALGORITHM = 'RS256'                  # the private key signs, the public key verifies

def issue_access_token(user):
    now = datetime.now(timezone.utc)
    payload = {
        'sub': str(user.id), 'tid': str(user.tenant_id),
        'roles': user.role_names,
        'iat': now, 'exp': now + timedelta(minutes=15),   # short lifetime
        'iss': 'auth.example.com', 'aud': 'api.example.com',
        'jti': str(uuid.uuid4()),                         # for revocation
    }
    return jwt.encode(payload, PRIVATE_KEY, algorithm=ALGORITHM)

def verify(token):
    claims = jwt.decode(token, PUBLIC_KEY,
                        algorithms=[ALGORITHM],          # NEVER from the header
                        audience='api.example.com',
                        issuer='auth.example.com')       # also checks exp
    if redis.exists(f'revoked:{claims["jti"]}'):
        raise jwt.InvalidTokenError('revoked')
    return claims

# ======================================================================
# Q: How do you organize a large Flask application?
#
# An application factory (create_app) so you can have different config in tests
# and production, Blueprints per domain (like Django apps), and extensions
# (Flask-SQLAlchemy, Flask-Migrate) that call init_app inside the factory.
# Business logic goes in a service layer, not in the views, so that a Celery
# task can call it too.
# ======================================================================

def create_app(config_object='app.config.Production'):
    app = Flask(__name__)
    app.config.from_object(config_object)
    db.init_app(app)
    migrate.init_app(app, db)
    from app.assets.routes import bp as assets_bp
    app.register_blueprint(assets_bp, url_prefix='/v1/assets')
    return app