ef806f8fbb
Phase 1-3 of decoupling plan: - Shared DB, models, infrastructure, browser, config, utils - Event infrastructure (domain_events outbox, bus, processor) - Structured logging - Generic container concept (container_type/container_id) - Alembic migrations for all schema changes Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
59 lines
2.2 KiB
Python
59 lines
2.2 KiB
Python
|
|
def register(app):
|
|
import json
|
|
from typing import Any
|
|
|
|
def _decode_headers(scope) -> dict[str, str]:
|
|
out = {}
|
|
for k, v in scope.get("headers", []):
|
|
try:
|
|
out[k.decode("latin1")] = v.decode("latin1")
|
|
except Exception:
|
|
out[repr(k)] = repr(v)
|
|
return out
|
|
|
|
def _safe(obj: Any):
|
|
# make scope json-serialisable; fall back to repr()
|
|
try:
|
|
json.dumps(obj)
|
|
return obj
|
|
except Exception:
|
|
return repr(obj)
|
|
|
|
class ScopeDumpMiddleware:
|
|
def __init__(self, app, *, log_bodies: bool = False):
|
|
self.app = app
|
|
self.log_bodies = log_bodies # keep False; bodies aren't needed for routing
|
|
|
|
async def __call__(self, scope, receive, send):
|
|
if scope["type"] in ("http", "websocket"):
|
|
# Build a compact view of keys relevant to routing
|
|
scope_view = {
|
|
"type": scope.get("type"),
|
|
"asgi": scope.get("asgi"),
|
|
"http_version": scope.get("http_version"),
|
|
"scheme": scope.get("scheme"),
|
|
"method": scope.get("method"),
|
|
"server": scope.get("server"),
|
|
"client": scope.get("client"),
|
|
"root_path": scope.get("root_path"),
|
|
"path": scope.get("path"),
|
|
"raw_path": scope.get("raw_path").decode("latin1") if scope.get("raw_path") else None,
|
|
"query_string": scope.get("query_string", b"").decode("latin1"),
|
|
"headers": _decode_headers(scope),
|
|
}
|
|
|
|
print("\n=== ASGI SCOPE (routing) ===")
|
|
print(json.dumps({_safe(k): _safe(v) for k, v in scope_view.items()}, indent=2))
|
|
print("=== END SCOPE ===\n", flush=True)
|
|
|
|
return await self.app(scope, receive, send)
|
|
|
|
# wrap LAST so you see what hits Quart
|
|
#app.asgi_app = ScopeDumpMiddleware(app.asgi_app)
|
|
|
|
|
|
from hypercorn.middleware import ProxyFixMiddleware
|
|
# trust a single proxy hop; use legacy X-Forwarded-* headers
|
|
app.asgi_app = ProxyFixMiddleware(app.asgi_app, mode="legacy", trusted_hops=1)
|