api: add better response documentation

This commit is contained in:
lilly 2026-05-07 21:38:22 +02:00
commit 745dfaf19f
Signed by: lilly
SSH key fingerprint: SHA256:y9T5GFw2A20WVklhetIxG1+kcg/Ce0shnQmbu1LQ37g
4 changed files with 97 additions and 16 deletions

View file

@ -1,16 +1,16 @@
from typing import Optional
from typing import Optional, List
import logging
import secrets
import sys
from datetime import datetime, UTC
from fastapi import FastAPI, Request, Response
from fastapi import FastAPI, Request, Response, HTTPException, status
from fastapi.responses import RedirectResponse
from contextlib import asynccontextmanager
from simple_openid_connect.client import OpenidClient
from simple_openid_connect.data import TokenSuccessResponse
from cachetools import TTLCache
from dooris_api import deps
from dooris_api.models import UserStatus, UserInfo
from dooris_api import deps, models, exceptions
logger = logging.getLogger(__name__)
@ -40,23 +40,24 @@ app = FastAPI(
title="Dooris", summary="API for interacting with HomeMatic door locks in CCCHH", docs_url="/api/docs",
lifespan=lifespan,
)
app.add_exception_handler(exceptions.HttpProblemException, exceptions.problem_exception_handler)
@app.get("/api/user-info/", name="get-user-info")
async def get_user_info(req: Request, current_user: deps.CurrentUser) -> UserStatus:
@app.get("/api/user-info/", name="get-user-info", tags=["auth"], responses={status.HTTP_401_UNAUTHORIZED: {"model": models.HttpProblemDetail}})
async def get_user_info(req: Request, current_user: deps.CurrentUser) -> models.UserStatus:
if current_user is None:
return UserStatus(is_logged_in=False, user_info=None)
return models.UserStatus(is_logged_in=False, user_info=None, guaranteed_session_until=None)
else:
return UserStatus(
return models.UserStatus(
is_logged_in=True,
guaranteed_session_until=datetime.fromtimestamp(current_user.id_token.exp, UTC),
user_info=UserInfo(
user_info=models.UserInfo(
username=current_user.id_token.preferred_username,
)
)
@app.get("/auth/login", response_class=RedirectResponse, status_code=302)
@app.get("/auth/login", tags=["auth"], response_class=RedirectResponse, status_code=302)
async def login_init(req: Request, resp: Response, oidc_client: deps.OpenidClient, next: Optional[str] = "") -> str:
logger.debug("starting user authentication with upstream identity provider")
@ -80,7 +81,7 @@ async def login_init(req: Request, resp: Response, oidc_client: deps.OpenidClien
return oidc_client.authorization_code_flow.start_authentication(state=state, nonce=nonce)
@app.get("/auth/login-callback", response_class=RedirectResponse, status_code=302)
@app.get("/auth/login-callback", tags=["auth"], response_class=RedirectResponse, status_code=302)
async def login_callback(req: Request, resp: Response, oidc_client: deps.OpenidClient):
# check that the user is currently in an authenticating state
# these cookies are set by the login_init() view
@ -109,4 +110,15 @@ async def login_callback(req: Request, resp: Response, oidc_client: deps.OpenidC
return f"/auth/login-error?{req.query_params}"
@app.get("/api/doors/", tags=["doors"], responses={status.HTTP_401_UNAUTHORIZED: {"model": models.HttpProblemDetail}})
async def list_doors(cache: deps.Cache, _user: deps.CurrentUser) -> List[models.Door]:
return [
models.Door(name="door1", description="A static door for testing", status="unknown"),
models.Door(name="door2", description="another static door for testing", status="unknown"),
]
@app.get("/api/doors/{name}", tags=["doors"], responses={status.HTTP_404_NOT_FOUND: {"model": models.HttpProblemDetail}, status.HTTP_401_UNAUTHORIZED: {"model": models.HttpProblemDetail}})
async def get_door(name: str, req: Request, cache: deps.Cache, _user: deps.CurrentUser) -> Optional[models.Door]:
raise exceptions.HttpProblemException(models.HttpProblemDetail.new_door_not_found(name, req.url))

View file

@ -4,8 +4,9 @@ from datetime import datetime, UTC, timedelta
from fastapi import Request, Depends, Response
from simple_openid_connect.data import TokenSuccessResponse
from simple_openid_connect.client import OpenidClient
from cachetools import Cache
from dooris_api import models
from dooris_api import models, exceptions
logger = logging.getLogger(__name__)
@ -40,7 +41,7 @@ async def get_current_user(req: Request, resp: Response, oidc_client: OpenidClie
# otherwise we can't meaningfully recover any user information or the user is simply not authenticated
logger.debug("no currently authenticated user")
return None
raise exceptions.HttpProblemException(models.HttpProblemDetail.new_unauthorized(req.url))
def persist_auth_state(oidc_client: OpenidClient, resp: Response, tokens: TokenSuccessResponse, auth_start_time: datetime, token_nonce: Optional[str] = None):

View file

@ -0,0 +1,26 @@
from typing import Mapping, Optional
from fastapi import HTTPException, Request, Response
from fastapi.responses import JSONResponse
from fastapi.utils import is_body_allowed_for_status_code
from fastapi.encoders import jsonable_encoder
from dooris_api import models
class HttpProblemException(HTTPException):
problem: models.HttpProblemDetail
headers: Optional[Mapping[str, str]]
def __init__(self, problem: models.HttpProblemDetail, headers: Optional[Mapping[str, str]] = None):
self.problem = problem
super().__init__(status_code=problem.status, headers=headers)
def __str__(self) -> str:
return str(self.problem)
async def problem_exception_handler(request: Request, exc: HttpProblemException) -> Response:
headers = exc.headers
if not is_body_allowed_for_status_code(exc.problem.status):
return Response(status_code=exc.status_code, headers=headers)
return JSONResponse(jsonable_encoder(exc.problem), status_code=exc.status_code, headers=exc.headers)

View file

@ -1,7 +1,37 @@
from typing import Optional
from typing import Optional, Self
from datetime import datetime
from pydantic import BaseModel
from pydantic import BaseModel, HttpUrl
from enum import Enum
from simple_openid_connect.data import IdToken
from starlette.datastructures import URL
from fastapi import status
class HttpProblemType(Enum):
"""
Statically known HTTP problem types using the [type URI scheme](https://datatracker.ietf.org/doc/rfc4151/)
"""
UNAUTHORIZED = "type:noc@hamburg.ccc.de,2026:UNAUTHORIZED"
DOOR_NOT_FOUND = "type:noc@hamburg.ccc.de,2026:DOOR_NOT_FOUND"
class HttpProblemDetail(BaseModel):
"""
API Error modeled after [RFC9475](https://www.rfc-editor.org/rfc/rfc9457.html).
"""
status: int
type: HttpProblemType
title: str
detail: str
instance: Optional[HttpUrl]
@classmethod
def new_unauthorized(cls, request_uri: str | URL) -> Self:
return cls(type=HttpProblemType.UNAUTHORIZED, status=status.HTTP_401_UNAUTHORIZED, title="Unauthorized", detail="You tried to access a ressource which requires authentication but you are not authenticated", instance=HttpUrl(str(request_uri)))
@classmethod
def new_door_not_found(cls, requested_door: str, request_uri: str | URL) -> Self:
return cls(type=HttpProblemType.DOOR_NOT_FOUND, status=status.HTTP_404_NOT_FOUND, title="Door not found", detail=f"You tried to interact with door {requested_door!r} that is not known to dooris", instance=str(request_uri))
class CurrentUser(BaseModel):
@ -14,6 +44,18 @@ class UserInfo(BaseModel):
class UserStatus(BaseModel):
is_logged_in: bool
guaranteed_session_until: datetime
guaranteed_session_until: Optional[datetime]
user_info: Optional[UserInfo]
class DoorStatus(Enum):
OPEN = "open"
CLOSED = "closed"
UNKNOWN = "unknown"
class Door(BaseModel):
name: str
description: str
status: DoorStatus