Backend Live Coding

# ======================================================================
# Q: Write a model and a query with SQLAlchemy 2.0 without N+1.
#
# lazy='raise' is the equivalent of a test with assertNumQueries: if someone
# forgets eager loading, they get an exception instead of 1,000 hidden queries.
# joinedload suits many-to-one (a single JOIN). selectinload suits one-to-many
# (a second query with WHERE id IN (...)), because a JOIN would multiply the
# rows.
# ======================================================================

from sqlalchemy import ForeignKey, String, select
from sqlalchemy.orm import (DeclarativeBase, Mapped, mapped_column,
                            relationship, selectinload, joinedload)

class Base(DeclarativeBase): ...

class Customer(Base):
    __tablename__ = 'customers'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(200))
    orders: Mapped[list['Order']] = relationship(back_populates='customer',
                                                 lazy='raise')   # N+1 -> exception

class Order(Base):
    __tablename__ = 'orders'
    id: Mapped[int] = mapped_column(primary_key=True)
    customer_id: Mapped[int] = mapped_column(ForeignKey('customers.id'), index=True)
    total: Mapped[int]                                   # in cents
    customer: Mapped[Customer] = relationship(back_populates='orders')

stmt = (select(Customer)
        .options(selectinload(Customer.orders))         # 2 queries in total
        .where(Customer.name.ilike('a%'))
        .order_by(Customer.id))
customers = session.scalars(stmt).all()

# ======================================================================
# Q: How do you test logic that depends on the current time?
#
# Never a bare datetime.now() inside the logic: either I pass the time in as a
# parameter, or I "freeze" time in the test.
# ======================================================================

import time_machine

@time_machine.travel(datetime(2026, 10, 25, 1, 30, tzinfo=UTC))   # DST change in Europe
def test_slot_active_after_dst_change():
    screen = Screen(timezone='Europe/Athens')
    slot = Slot(weekdays=[6], start_time=time(4, 0), end_time=time(5, 0))
    assert is_slot_active(slot, screen, datetime.now(UTC))

# ======================================================================
# Q: How do you test Celery tasks?
#
# The task is a function: I call it directly (task.run(...) or task(...)) in the
# unit test. To check that it was scheduled correctly, I mock .delay. I avoid
# CELERY_TASK_ALWAYS_EAGER as a global setting because it hides problems
# (serialization, on_commit, retries).
# ======================================================================

@pytest.mark.django_db
def test_confirm_upload_enqueues_processing(api_client, ws_user, mocker,
                                            django_capture_on_commit_callbacks):
    media = MediaFactory(workspace=ws_user.workspace, status='pending')
    delay = mocker.patch('media.views.process_media.delay')
    api_client.force_authenticate(ws_user)

    with django_capture_on_commit_callbacks(execute=True):
        r = api_client.post(f'/api/v2/media/{media.pk}/confirm/')

    assert r.status_code == 202
    delay.assert_called_once_with(media.pk)

@pytest.mark.django_db
def test_other_workspace_cannot_confirm(api_client, ws_user):
    media = MediaFactory()                     # different workspace
    api_client.force_authenticate(ws_user)
    r = api_client.post(f'/api/v2/media/{media.pk}/confirm/')
    assert r.status_code == 404

# ======================================================================
# Q: How do you test code that calls an external HTTP API?
#
# In unit tests I never go out to the internet. With the responses library (for
# requests) or respx (for httpx) I define what the "API" returns and I also test
# the bad scenarios: 429, 500, timeout, malformed JSON.
# For integration: a few tests against a real provider sandbox, in a separate CI
# job (not on every commit), or recorded responses with VCR.py / pytest-
# recording.
# ======================================================================

import responses

@responses.activate
def test_forecast_raises_rate_limited():
    responses.get('https://api.example.com/v2/forecast',
                  status=429, headers={'Retry-After': '30'})
    with pytest.raises(RateLimited) as exc:
        WeatherClient('key').forecast('Athens')
    assert exc.value.retry_after == 30

@responses.activate
def test_forecast_timeout_is_set():
    responses.get('https://api.example.com/v2/forecast',
                  body=requests.exceptions.ConnectTimeout())
    with pytest.raises(requests.exceptions.ConnectTimeout):
        WeatherClient('key').forecast('Athens')

# ======================================================================
# Q: An endpoint is slow in production. Describe step by step how you find the
# cause.
#
# - Measure first: APM (Sentry Performance, New Relic, Datadog) to see where the
#   time goes: database, external APIs, Python.
# - Locally with similar data: django-debug-toolbar or django-silk for the
#   number and duration of queries, and duplicate queries (N+1).
# - For the slow query: qs.explain(analyze=True, buffers=True) and
#   pg_stat_statements. I look for a Seq Scan on a large table, wrong row
#   estimates, a Sort on disk.
# - If the time is in Python: cProfile / py-spy (py-spy also works on a running
#   process, with no code change).
# - Fix it (index, select_related, cache, move to Celery) and add a test with
#   assertNumQueries so the problem does not come back.
# ======================================================================

def test_screen_list_queries(api_client, ws_user, django_assert_max_num_queries):
    ScreenFactory.create_batch(30, workspace=ws_user.workspace)
    api_client.force_authenticate(ws_user)
    with django_assert_max_num_queries(5):     # constant, independent of the count
        r = api_client.get('/api/v2/screens/')
    assert r.status_code == 200