Coverage for polar/auth/routing.py: 100%

1import inspect 

2import typing 

3from collections.abc import Callable 

4 

5from fastapi.params import Depends 

6from fastapi.routing import APIRoute 

7 

8from polar.auth.dependencies import _Authenticator 

9from polar.auth.scope import RESERVED_SCOPES 

10 

11 

12class DocumentedAuthSubjectAPIRoute(APIRoute): 

13 """ 

14 A subclass of `APIRoute` that automatically 

15 documents the allowed subjects and scopes for the endpoint. 

16 """ 

17 

18 def __init__( 

19 self, path: str, endpoint: Callable[..., typing.Any], **kwargs: typing.Any 

20 ) -> None: 

21 openapi_extra = kwargs.get("openapi_extra") or {} 

22 # Check we haven't already added the allowed subjects 

23 if "x-polar-allowed-subjects" not in openapi_extra: 

24 for param in typing.get_type_hints(endpoint, include_extras=True).values(): 

25 if typing.get_origin(param) is not typing.Annotated: 

26 continue 

27 

28 metadata = param.__metadata__ 

29 if len(metadata) == 0 or not isinstance(metadata[0], Depends): 

30 continue 

31 

32 dependency = metadata[0].dependency 

33 if not isinstance(dependency, _Authenticator): 

34 continue 

35 

36 allowed_subjects = dependency.allowed_subjects 

37 required_scopes = dependency.required_scopes 

38 

39 allowed_subjects_names = sorted( 

40 [allowed_subject.__name__ for allowed_subject in allowed_subjects] 

41 ) 

42 

43 kwargs["openapi_extra"] = { 

44 "x-polar-allowed-subjects": allowed_subjects_names, 

45 **openapi_extra, 

46 } 

47 

48 description = kwargs["description"] or inspect.cleandoc( 

49 endpoint.__doc__ or "" 

50 ) 

51 scopes_list = [ 

52 f"`{s}`" 

53 for s in sorted(required_scopes or []) 

54 if s not in RESERVED_SCOPES 

55 ] 

56 if scopes_list: 

57 description += f"\n\n**Scopes**: {' '.join(scopes_list)}" 

58 kwargs["description"] = description 

59 

60 break 

61 

62 super().__init__(path, endpoint, **kwargs) 

63 

64 

65__all__ = ["DocumentedAuthSubjectAPIRoute"]