"""Falcon framework integration.

`RahmFalconMiddleware` (WSGI) and `RahmFalconAsyncMiddleware` (ASGI) open a

rahm scope per request — binding `trace_id`, `http_method`, `http_path` — and

emit an `http_request_completed` access entry on the way out.

`rahm_error_handler` / `rahm_error_handler_async` registered against

`Exception` turn truly uncaught Python exceptions into an ERROR entry (full

stack + request context) plus a JSON 500. Falcon's built-in handlers for

`HTTPError` and `HTTPStatus` are more specific, so they keep handling 4xx/3xx

without going through this handler.

Wiring (WSGI):

import falcon

from rahm.falcon import RahmFalconMiddleware, rahm_error_handler

app = falcon.App(middleware=[RahmFalconMiddleware()])

app.add_error_handler(Exception, rahm_error_handler)

Wiring (ASGI):

import falcon.asgi

from rahm.falcon import RahmFalconAsyncMiddleware, rahm_error_handler_async

app = falcon.asgi.App(middleware=[RahmFalconAsyncMiddleware()])

app.add_error_handler(Exception, rahm_error_handler_async)

"""

import logging

import os

import time

import falcon

import rahm

from rahm._common import ulid as _ulid, parse_trace_id_setting, parse_quiet_paths, is_quiet_access

from rahm.log import LocalFormatter, JsonFormatter, LogfmtFormatter

_TRACE_ID_ENABLED = parse_trace_id_setting()

_QUIET_PATHS = parse_quiet_paths()

# --- per-request scope: opened in process_request, closed in process_response ---

def _open_scope(req):

fields = {

'http_method': req.method,

'http_path': req.path,

}

trace_id = None

if _TRACE_ID_ENABLED:

trace_id = req.get_header('X-Trace-Id') or _ulid()

fields['trace_id'] = trace_id

scope_cm = rahm.log.scope(**fields)

scope_cm.__enter__()

req.context.rahm_scope = scope_cm

req.context.rahm_trace_id = trace_id

req.context.rahm_start = time.monotonic()

req.context.rahm_skip_access = False

def _close_scope(req):

req.context.rahm_scope.__exit__(None, None, None)

def _status_code(resp):

status = resp.status

if isinstance(status, int):

return status

return int(str(status).split(' ', 1)[0])

def _emit_access(req, resp):

duration_ms = round((time.monotonic() - req.context.rahm_start) * 1000, 3)

status = _status_code(resp)

emit = rahm.log.debug if is_quiet_access(req.path, status, _QUIET_PATHS) else rahm.log.info

emit(

'http_request_completed',

f'{req.method} {req.path} {status}',

http_status=status,

http_duration_ms=duration_ms,

)

def _finalize_response(req, resp):

try:

if not req.context.rahm_skip_access:

_emit_access(req, resp)

if _TRACE_ID_ENABLED and req.context.rahm_trace_id:

resp.set_header('X-Trace-Id', req.context.rahm_trace_id)

finally:

_close_scope(req)

# --- request context for error logging ---

def _headers_block(req):

out = ''

for k, v in req.headers.items():

name = k.lower()

if name == 'cookie':

out += 'cookie: \n'

for cookie in [s.strip() for s in v.split(';')]:

cname, _, cvalue = cookie.partition('=')

out += ' name: ' + cname + '\n'

out += ' value: ' + cvalue + '\n'

else:

out += name + ': ' + v + '\n'

return out.rstrip('\n')

def _query_block(req):

query = req.query_string or ''

if not query:

return ''

out = ''

for param in query.split('&'):

k, _, v = param.partition('=')

out += k + ': ' + v + '\n'

return out.rstrip('\n')

def _client(req):

return req.remote_addr or ''

def _decode_body(body):

if not body:

return ''

try:

return body.decode('utf-8')

except UnicodeDecodeError:

return f'<binary {len(body)} bytes>'

def _wsgi_body(req):

try:

body = req.bounded_stream.read()

except Exception:

return '<unavailable>'

return _decode_body(body)

async def _asgi_body(req):

try:

body = await req.stream.read()

except Exception:

return '<unavailable>'

return _decode_body(body)

def _wsgi_request_context(req):

return {

'http_client': _client(req),

'http_headers': _headers_block(req),

'http_query_params': _query_block(req),

'http_body': _wsgi_body(req),

}

async def _asgi_request_context(req):

return {

'http_client': _client(req),

'http_headers': _headers_block(req),

'http_query_params': _query_block(req),

'http_body': await _asgi_body(req),

}

# --- middlewares ---

class RahmFalconMiddleware:

"""WSGI: opens a per-request scope and emits the access log on the way out."""

def process_request(self, req, resp):

_open_scope(req)

def process_response(self, req, resp, resource, req_succeeded):

_finalize_response(req, resp)

class RahmFalconAsyncMiddleware:

"""ASGI counterpart of RahmFalconMiddleware."""

async def process_request(self, req, resp):

_open_scope(req)

async def process_response(self, req, resp, resource, req_succeeded):

_finalize_response(req, resp)

# --- error handlers ---

# Falcon registers default handlers for HTTPError/HTTPStatus that are more

# specific than Exception, so this only catches true uncaught Python errors.

# We log the error and signal the middleware to skip its own access log so

# each failed request produces exactly one entry, matching the Starlette story.

def _log_uncaught(req, ex, ctx):

duration_ms = round((time.monotonic() - req.context.rahm_start) * 1000, 3)

req.context.rahm_skip_access = True

rahm.log.error(

'uncaught_exception',

f'Uncaught exception - {ex}',

exc_info=(type(ex), ex, ex.__traceback__),

http_status=500,

http_duration_ms=duration_ms,

**ctx,

)

def _render_500(resp):

resp.status = falcon.HTTP_500

resp.content_type = falcon.MEDIA_JSON

resp.text = '"internal server error"'

def rahm_error_handler(req, resp, ex, params):

_log_uncaught(req, ex, _wsgi_request_context(req))

_render_500(resp)

async def rahm_error_handler_async(req, resp, ex, params):

_log_uncaught(req, ex, await _asgi_request_context(req))

_render_500(resp)

# --- granian log bridge ---

# Granian emits its own lifecycle/access lines through stdlib logging

# (loggers `_granian` and `granian.access`). Some records carry a template

# msg + args tuple; expand them so the rahm formatter (which reads record.msg

# directly) doesn't need to know about %-substitution. Granian's lines have no

# rahm event, so tag them to satisfy the mandatory-field rule.

class GranianNormalizer(logging.Filter):

def filter(self, record):

record.msg = record.getMessage()

record.args = ()

record.event = 'granian'

return True

# Pass to Granian's `log_dictconfig=` (or dump to JSON for `--log-config`) to

# route Granian's own logs through rahm instead of suppressing them with

# `log_enabled=False`. Mirrors rahm.starlette.uvicorn_log_config: class objects

# (not dotted strings) because rahm/__init__.py rebinds rahm.log to the Logger

# instance, so 'rahm.log.X' wouldn't resolve at server boot.

def granian_log_config():

formatter = {

'text': LocalFormatter,

'logfmt': LogfmtFormatter,

}.get(os.environ.get('RAHM_LOG_FORMAT', 'json').lower(), JsonFormatter)

return {

'version': 1,

'disable_existing_loggers': False,

'formatters': {'rahm': {'()': formatter}},

'filters': {'granian_normalizer': {'()': GranianNormalizer}},

'handlers': {

'rahm': {

'class': 'logging.StreamHandler',

'formatter': 'rahm',

'filters': ['granian_normalizer'],

},

},

'loggers': {

'_granian': {'handlers': ['rahm'], 'level': 'INFO', 'propagate': False},

'granian.access': {'handlers': ['rahm'], 'level': 'INFO', 'propagate': False},

},

}

# --- gunicorn log bridge ---

# Falcon's WSGI middleware/error-handler are server-agnostic, so they run under

# gunicorn (sync, gthread, gevent workers) unchanged. This bridge only routes

# gunicorn's *own* logs through rahm. Like Granian, gunicorn emits template msg

# + args records (e.g. "Booting worker with pid: %s"); expand them so the rahm

# formatter can read record.msg directly, and tag them so they carry an event.

class GunicornNormalizer(logging.Filter):

def filter(self, record):

record.msg = record.getMessage()

record.args = ()

record.event = 'gunicorn'

return True

# Set as `logconfig_dict` in your gunicorn config (gunicorn.conf.py:

# `logconfig_dict = gunicorn_log_config()`) to route gunicorn's lifecycle logs

# (`gunicorn.error`) through rahm. gunicorn shallow-merges this over its

# defaults, so re-declaring formatters/handlers/loggers replaces its built-in

# text formatter entirely. `gunicorn.access` is deliberately silenced (no

# handlers, no propagation): setting logconfig_dict turns gunicorn's access log

# on, but the Falcon middleware already emits one http_request_completed entry

# per request, so letting it through would double-log every request. Class

# objects, not dotted strings, for the same reason as granian_log_config. gunicorn merges

# this over its own CONFIG_DEFAULTS, so `root` is overridden too — otherwise the

# default root would still reference gunicorn's `console` handler we replace.

def gunicorn_log_config():

formatter = {

'text': LocalFormatter,

'logfmt': LogfmtFormatter,

}.get(os.environ.get('RAHM_LOG_FORMAT', 'json').lower(), JsonFormatter)

return {

'version': 1,

'disable_existing_loggers': False,

'formatters': {'rahm': {'()': formatter}},

'filters': {'gunicorn_normalizer': {'()': GunicornNormalizer}},

'handlers': {

'rahm': {

'class': 'logging.StreamHandler',

'formatter': 'rahm',

'filters': ['gunicorn_normalizer'],

},

},

'root': {'level': 'INFO', 'handlers': []},

'loggers': {

'gunicorn.error': {'handlers': ['rahm'], 'level': 'INFO', 'propagate': False},

'gunicorn.access': {'handlers': [], 'level': 'INFO', 'propagate': False},

},

}