Production logging with request IDs propagated from the API to Celery tasks

# ======================================================================
# Q: How do you set up logging so you can find a problem in production?
#
# Structured JSON logs with a request_id/correlation id that passes from the API
# to Celery tasks and to calls to other services. That way I filter all the logs
# of one request in one place (ELK, Loki, CloudWatch).
# I never write passwords, tokens or personal data (GDPR) to logs. I add a
# filter that masks them.
# ======================================================================

import contextvars, logging, uuid
request_id_var = contextvars.ContextVar('request_id', default='-')

class RequestIdFilter(logging.Filter):
    def filter(self, record):
        record.request_id = request_id_var.get()
        return True

@app.middleware('http')
async def add_request_id(request, call_next):
    rid = request.headers.get('X-Request-ID') or uuid.uuid4().hex
    token = request_id_var.set(rid)
    try:
        response = await call_next(request)
    finally:
        request_id_var.reset(token)
    response.headers['X-Request-ID'] = rid
    return response

# Celery: pass the id in the task headers
process_import.apply_async(args=[import_id], headers={'request_id': request_id_var.get()})