"""Consumer-facing API documentation.
`GET /openapi.json` returns a merged OpenAPI document scoped to the calling
API key: every operation the key holds a grant on, across all services,
rewritten to the gateway's own routes (`/{slug}/...`). `GET /docs` serves a
Swagger UI portal around it.
"""
import re
from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse, JSONResponse
from sqlalchemy.orm import Session
from app import config, discovery, security
from app.database import get_db
from app.models import ApiKey, Endpoint, Service
from app.proxy import _error, _pattern_to_regex
router = APIRouter()
_HTTP_METHODS = ("get", "post", "put", "patch", "delete", "head", "options")
_REF_RE = re.compile(r"^#/components/([A-Za-z]+)/(.+)$")
def _rewrite_refs(obj, slug: str):
"""Namespace `$ref`s into a service's renamed components (atlas_Error)."""
if isinstance(obj, dict):
out = {}
for key, value in obj.items():
if key == "$ref" and isinstance(value, str):
m = _REF_RE.match(value)
out[key] = (f"#/components/{m.group(1)}/{slug}_{m.group(2)}"
if m else value)
else:
out[key] = _rewrite_refs(value, slug)
return out
if isinstance(obj, list):
return [_rewrite_refs(item, slug) for item in obj]
return obj
def _grant_matches(grant: Endpoint, method: str, path: str) -> bool:
return (grant.method in ("*", method)
and _pattern_to_regex(grant.path).match(path) is not None)
def _service_paths(service: Service, grants: list[Endpoint], spec: dict | None) -> dict:
"""The slug-prefixed path entries of one service, filtered to the grants.
Operations keep their upstream tags — they become the second grouping
level; untagged operations fall into 'General'."""
slug = service.slug
if spec:
included = {}
for path, operations in spec.get("paths", {}).items():
if not isinstance(operations, dict):
continue
kept = {}
for method, op in operations.items():
if method.lower() not in _HTTP_METHODS or not isinstance(op, dict):
continue
if any(_grant_matches(g, method.upper(), path) for g in grants):
kept[method] = {**op, "tags": list(op.get("tags") or ["General"])}
if kept:
entry = {k: v for k, v in operations.items() if k.lower() not in _HTTP_METHODS}
entry.update(kept)
included[f"/{slug}{path}"] = entry
return _rewrite_refs(included, slug)
# No OpenAPI document upstream — synthesize minimal entries from the catalog.
included = {}
for g in grants:
methods = [g.method.lower()] if g.method != "*" else ["get", "post", "put", "patch", "delete"]
entry = included.setdefault(f"/{slug}{g.path}", {})
for method in methods:
entry[method] = {
"summary": g.description or ("(any method)" if g.method == "*" else ""),
"tags": ["General"],
"responses": {"default": {"description": "Proxied upstream response"}},
}
return included
def build_key_spec(db: Session, api_key: ApiKey, gateway_url: str) -> dict:
"""Two-level structure: each service is an `x-tagGroups` entry (first
level, with its description in the intro), containing the upstream's own
operation tags (second level). Tag names shared by several services are
disambiguated with the service name."""
by_service: dict[int, list[Endpoint]] = {}
for grant in api_key.endpoints:
by_service.setdefault(grant.service_id, []).append(grant)
merged: list[tuple[Service, dict]] = [] # (service, its filtered paths)
used_tags: dict[int, list[str]] = {} # service id -> upstream tags, in order
tag_descriptions: dict[tuple[int, str], str] = {}
upstream_infos: dict[int, dict] = {} # service id -> upstream spec's `info`
components: dict = {}
ordered = sorted(by_service.values(), key=lambda g: g[0].service.name)
for grants in ordered:
service = grants[0].service
if not service.is_active:
continue
spec = discovery.get_spec(db, service)
paths = _service_paths(service, grants, spec)
if not paths:
continue
order: list[str] = []
for entry in paths.values():
for method, op in entry.items():
if method.lower() in _HTTP_METHODS and isinstance(op, dict):
for tag in op.get("tags", []):
if tag not in order:
order.append(tag)
used_tags[service.id] = order
if spec:
info = spec.get("info")
upstream_infos[service.id] = info if isinstance(info, dict) else {}
for tag in spec.get("tags") or []:
if isinstance(tag, dict) and tag.get("name") in order:
tag_descriptions[(service.id, tag["name"])] = tag.get("description", "")
for section, items in spec.get("components", {}).items():
if section == "securitySchemes" or not isinstance(items, dict):
continue
renamed = {f"{service.slug}_{name}": schema for name, schema in items.items()}
components.setdefault(section, {}).update(_rewrite_refs(renamed, service.slug))
merged.append((service, paths))
# A tag name used by several services must stay unique in the flat tag list.
tag_counts: dict[str, int] = {}
for names in used_tags.values():
for name in names:
tag_counts[name] = tag_counts.get(name, 0) + 1
def final_name(service: Service, tag: str) -> str:
return tag if tag_counts[tag] == 1 else f"{tag} ({service.name})"
paths_out: dict = {}
tags_out: list[dict] = []
tag_groups: list[dict] = []
intro = [f"Operations available to API key **{api_key.name}**. "
f"Send the key in the `{config.API_KEY_HEADER}` header."]
for service, paths in merged:
finals = []
for tag in used_tags[service.id]:
name = final_name(service, tag)
finals.append(name)
tags_out.append({"name": name,
"description": tag_descriptions.get((service.id, tag), "")})
for entry in paths.values():
for method, op in entry.items():
if method.lower() in _HTTP_METHODS and isinstance(op, dict):
op["tags"] = [final_name(service, t) for t in op.get("tags", ["General"])]
paths_out.update(paths)
tag_groups.append({"name": service.name, "tags": finals})
info = upstream_infos.get(service.id, {})
section = f"## {service.name}\n\n"
title = info.get("title", "")
if title and title.lower() != service.name.lower():
section += f"*{title}*\n\n"
# The admin-entered description overrides the upstream's own.
description = service.description or info.get("description") or ""
if description:
section += description.strip() + "\n\n"
section += f"Routes under `/{service.slug}/…`"
intro.append(section)
components["securitySchemes"] = {
"ApiKeyAuth": {"type": "apiKey", "in": "header", "name": config.API_KEY_HEADER},
}
return {
"openapi": "3.0.3",
"info": {
"title": "API Gateway",
"version": "1.0.0",
"description": "\n\n".join(intro),
},
"servers": [{"url": gateway_url}],
"tags": tags_out,
"x-tagGroups": tag_groups,
"paths": dict(sorted(paths_out.items())),
"components": components,
"security": [{"ApiKeyAuth": []}],
}
@router.get("/openapi.json", include_in_schema=False)
def key_scoped_openapi(request: Request, db: Session = Depends(get_db)):
plain_key = request.headers.get(config.API_KEY_HEADER)
if not plain_key:
return _error(401, "missing_api_key",
f"Provide your API key in the {config.API_KEY_HEADER} header.")
api_key = (db.query(ApiKey)
.filter(ApiKey.key_hash == security.hash_api_key(plain_key))
.one_or_none())
if api_key is None or not api_key.is_active or not api_key.user.is_active:
return _error(403, "invalid_api_key", "API key is unknown or has been revoked.")
spec = build_key_spec(db, api_key, str(request.base_url).rstrip("/"))
return JSONResponse(spec, headers={"Cache-Control": "no-store"})
_DOCS_PAGE = """
API Gateway · Docs
API Gateway
Enter your API key above to see the endpoints available to you.