Coverage for polar/auth/routing.py: 100%
32 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:52 +0000
1import inspect 1a
2import typing 1a
3from collections.abc import Callable 1a
5from fastapi.params import Depends 1a
6from fastapi.routing import APIRoute 1a
8from polar.auth.dependencies import _Authenticator 1a
9from polar.auth.scope import RESERVED_SCOPES 1a
12class DocumentedAuthSubjectAPIRoute(APIRoute): 1a
13 """
14 A subclass of `APIRoute` that automatically
15 documents the allowed subjects and scopes for the endpoint.
16 """
18 def __init__( 1a
19 self, path: str, endpoint: Callable[..., typing.Any], **kwargs: typing.Any
20 ) -> None:
21 openapi_extra = kwargs.get("openapi_extra") or {} 1ab
22 # Check we haven't already added the allowed subjects
23 if "x-polar-allowed-subjects" not in openapi_extra: 1ab
24 for param in typing.get_type_hints(endpoint, include_extras=True).values(): 1ab
25 if typing.get_origin(param) is not typing.Annotated: 1ab
26 continue 1ab
28 metadata = param.__metadata__ 1ab
29 if len(metadata) == 0 or not isinstance(metadata[0], Depends): 1ab
30 continue 1ab
32 dependency = metadata[0].dependency 1ab
33 if not isinstance(dependency, _Authenticator): 1ab
34 continue 1ab
36 allowed_subjects = dependency.allowed_subjects 1a
37 required_scopes = dependency.required_scopes 1a
39 allowed_subjects_names = sorted( 1a
40 [allowed_subject.__name__ for allowed_subject in allowed_subjects]
41 )
43 kwargs["openapi_extra"] = { 1a
44 "x-polar-allowed-subjects": allowed_subjects_names,
45 **openapi_extra,
46 }
48 description = kwargs["description"] or inspect.cleandoc( 1a
49 endpoint.__doc__ or ""
50 )
51 scopes_list = [ 1a
52 f"`{s}`"
53 for s in sorted(required_scopes or [])
54 if s not in RESERVED_SCOPES
55 ]
56 if scopes_list: 1a
57 description += f"\n\n**Scopes**: {' '.join(scopes_list)}" 1a
58 kwargs["description"] = description 1a
60 break 1a
62 super().__init__(path, endpoint, **kwargs) 1ab
65__all__ = ["DocumentedAuthSubjectAPIRoute"] 1a