Backend Live Coding

# ======================================================================
# Q: Write an endpoint that securely receives webhooks from an external service.
#
# (1) Verify the HMAC signature over the raw body, using compare_digest. (2)
# Check the timestamp for replay attacks, if the provider supplies one. (3)
# Dedup with a unique event_id, because providers deliver at-least-once. (4)
# Respond quickly with 200 and do the processing in Celery, otherwise the
# provider times out and resends. (5) Don't rely on the order of arrival of
# events.
# ======================================================================

import hashlib, hmac

@csrf_exempt
@require_POST
def billing_webhook(request):
    sig = request.headers.get('X-Signature', '')
    expected = hmac.new(settings.WEBHOOK_SECRET.encode(), request.body,
                        hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sig, expected):      # constant-time
        return HttpResponse(status=401)

    event = json.loads(request.body)
    _, created = WebhookEvent.objects.get_or_create(
        event_id=event['id'], defaults={'payload': event})   # unique -> dedup
    if created:
        transaction.on_commit(lambda: process_webhook.delay(event['id']))
    return HttpResponse(status=200)                 # fast 2xx

# ======================================================================
# Q: How do you refresh an OAuth2 access token without 50 workers doing it at
# the same time?
#
# I store the token and its expiry encrypted in the database. I refresh shortly
# before it expires. To avoid a double refresh (some providers invalidate the
# old refresh token), I lock the row with select_for_update and re-check inside
# the lock.
# If the refresh fails with invalid_grant (the user revoked access), I mark the
# integration as disconnected and notify the customer, instead of retrying
# forever.
# ======================================================================

def get_access_token(integration_id):
    integ = Integration.objects.get(pk=integration_id)
    if integ.expires_at > now() + timedelta(minutes=2):
        return integ.access_token                  # fast path
    with transaction.atomic():
        integ = Integration.objects.select_for_update().get(pk=integration_id)
        if integ.expires_at > now() + timedelta(minutes=2):
            return integ.access_token              # another worker already refreshed it
        tokens = oauth_provider.refresh(integ.refresh_token)
        integ.access_token = tokens['access_token']
        integ.refresh_token = tokens.get('refresh_token', integ.refresh_token)
        integ.expires_at = now() + timedelta(seconds=tokens['expires_in'])
        integ.save(update_fields=['access_token', 'refresh_token', 'expires_at'])
        return integ.access_token

# ======================================================================
# Q: How do you call an external API from Celery with retries?
#
# I pass an id and not a model instance (the object in the queue may be stale).
# The task is idempotent: if it runs twice, the result is the same. With
# acks_late, if the worker dies, the task runs again. That is why idempotency is
# mandatory.
# ======================================================================

@shared_task(bind=True,
             autoretry_for=(ExternalAPIError, requests.RequestException),
             retry_backoff=True,          # 1s, 2s, 4s, 8s ...
             retry_backoff_max=600,
             retry_jitter=True,
             max_retries=8,
             acks_late=True)
def refresh_weather_widget(self, widget_id):
    widget = Widget.objects.get(pk=widget_id)   # we pass an id, not an object
    try:
        data = WeatherClient(settings.WEATHER_KEY).forecast(widget.city)
    except RateLimited as exc:
        raise self.retry(exc=exc, countdown=exc.retry_after)
    Widget.objects.filter(pk=widget_id).update(payload=data, refreshed_at=now())

# ======================================================================
# Q: Write a production-ready client for an external API.
#
# What the interviewer looks for: always a timeout (requests waits forever by
# default), a Session for connection pooling, retry only on idempotent methods
# and on 5xx, respecting 429 Retry-After, our own exceptions so the rest of the
# code does not depend on requests, and never calling an external API inside a
# DB transaction.
# ======================================================================

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class ExternalAPIError(Exception): ...
class RateLimited(ExternalAPIError):
    def __init__(self, retry_after): self.retry_after = retry_after

class WeatherClient:
    BASE_URL = 'https://api.example.com/v2'

    def __init__(self, api_key, timeout=(3.05, 10)):   # (connect, read)
        self.session = requests.Session()                # connection pooling
        retry = Retry(total=3, backoff_factor=0.5,
                      status_forcelist=[502, 503, 504],
                      allowed_methods=['GET'])          # idempotent only
        self.session.mount('https://', HTTPAdapter(max_retries=retry))
        self.session.headers['Authorization'] = f'Bearer {api_key}'
        self.timeout = timeout

    def forecast(self, city):
        r = self.session.get(f'{self.BASE_URL}/forecast',
                             params={'q': city}, timeout=self.timeout)
        if r.status_code == 429:
            raise RateLimited(int(r.headers.get('Retry-After', 60)))
        if not r.ok:
            raise ExternalAPIError(f'{r.status_code}: {r.text[:200]}')
        return r.json()

# ======================================================================
# Q: Pagination on a list with millions of records (e.g. proof-of-play logs)?
#
# Offset pagination (LIMIT/OFFSET) becomes slow on large pages, because the
# database reads and discards all the previous rows, and it shows duplicate or
# missing records when new ones are inserted. I use CursorPagination (keyset)
# with a stable ordering on an indexed field, and avoid COUNT(*) on huge tables.
# ======================================================================

class PlayLogPagination(CursorPagination):
    page_size = 100
    ordering = '-id'          # unique and indexed

# Produces: WHERE id < :last_seen_id ORDER BY id DESC LIMIT 101