Backend Live Coding

# ======================================================================
# Q: A Celery worker or gunicorn process is stuck at 100% CPU or not responding.
# What do you do?
#
# py-spy is a sampling profiler that reads the process memory from the outside,
# so it is safe in production. The dump immediately shows whether the process is
# waiting on a socket without a timeout (the usual culprit: requests without a
# timeout to the ERP), on a database lock, or in an infinite loop. Once I have
# the stack, I restart as mitigation.
# ======================================================================

# stack of all threads without a restart and without changing code
py-spy dump --pid 12345

# live top-like view: which functions eat CPU
py-spy top --pid 12345

# flamegraph for 30s
py-spy record -o profile.svg --pid 12345 --duration 30

# in a container SYS_PTRACE is required
docker exec -it worker py-spy dump --pid 1

# ======================================================================
# Q: How do you use pdb? Give a few commands.
#
# pytest --pdb is the most useful: it drops you into the failing frame with all
# the local variables. In production I never do interactive debugging; there I
# work with logs, Sentry and the tools that follow.
# ======================================================================

def apply_discount(order):
    breakpoint()                  # PYTHONBREAKPOINT=0 disables it

(Pdb) p order.total, order.customer.discount_rate
(Pdb) pp vars(order)
(Pdb) n / s / c               # next, step into, continue
(Pdb) w                       # stack trace: who called me
(Pdb) u / d                   # move up/down the stack
(Pdb) b pricing.py:42, qty > 100   # conditional breakpoint
(Pdb) interact                # full Python REPL in the current frame

# post-mortem on a failing test
pytest tests/test_pricing.py -x --pdb

# ======================================================================
# Q: How do you reproduce locally a bug that appears only in production?
#
# Many bugs are about data or concurrency: legacy records from ASP with NULLs or
# odd encodings, two workers running at the same time. So: the same PostgreSQL
# version, an anonymized dump, a real Celery worker (not eager mode, which hides
# race conditions and on_commit). For step-by-step debugging I put in
# breakpoint() or attach the IDE through debugpy.
# ======================================================================

# docker-compose.debug.yml
services:
  db:
    image: postgres:16                 # same major version as production
    environment: { POSTGRES_PASSWORD: dev }
    volumes: ["./dumps:/dumps"]
  redis:
    image: redis:7
  web:
    build: .
    environment:
      DATABASE_URL: postgres://postgres:dev@db/app
      CELERY_TASK_ALWAYS_EAGER: "false"
    ports: ["8000:8000", "5678:5678"]  # 5678 for debugpy
    command: python -m debugpy --listen 0.0.0.0:5678 manage.py runserver 0.0.0.0:8000
  worker:
    build: .
    command: celery -A app worker -l info --concurrency 4

# anonymized dump from staging
# pg_restore -d app --no-owner /dumps/staging_anon.dump

# ======================================================================
# Q: How do you create test data without huge fixture files?
#
# With factory_boy each test declares only what it cares about, and the rest
# have sensible defaults. django_assert_num_queries pins the number of queries
# so I catch N+1 regressions in CI.
# ======================================================================

import factory
from factory.django import DjangoModelFactory

class CustomerFactory(DjangoModelFactory):
    class Meta:
        model = Customer
    name = factory.Faker('company', locale='el_GR')
    vat_number = factory.Sequence(lambda n: f'EL{n:09d}')
    kind = 'b2b'

class OrderFactory(DjangoModelFactory):
    class Meta:
        model = Order
    customer = factory.SubFactory(CustomerFactory)
    status = 'placed'

    class Params:
        paid = factory.Trait(status='paid')

def test_report_counts_only_paid(django_assert_num_queries):
    OrderFactory.create_batch(3, paid=True)
    OrderFactory()                                   # placed
    with django_assert_num_queries(1):
        assert revenue_report()['orders'] == 3

# ======================================================================
# Q: How do you mock an external HTTP API (e.g. a courier or payment gateway)?
#
# I mock at the HTTP boundary, not my own function, so parsing, timeouts and
# retry are tested too. I always test the unhappy paths: 5xx, timeout, invalid
# JSON, 429 with Retry-After. Most production bugs in the e-shops came from
# there.
# ======================================================================

# httpx -> respx
import httpx, respx

@respx.mock
def test_courier_label_retry_on_503():
    route = respx.post('https://api.courier.gr/v2/labels').mock(side_effect=[
        httpx.Response(503),
        httpx.Response(201, json={'tracking': 'GR123'}),
    ])
    assert create_label(order_id=5) == 'GR123'
    assert route.call_count == 2

# requests -> responses
import responses

@responses.activate
def test_payment_declined():
    responses.post('https://pay.example.com/charges',
                   json={'status': 'declined', 'code': 'insufficient_funds'}, status=402)
    with pytest.raises(PaymentDeclined):
        charge(order_id=5, amount_cents=1999)