Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/_base/controller.py: 76%
113 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 15:48 +0000
1"""
2This file contains code taken from fastapi-utils project. The code is licensed under the MIT license.
4See their repository for details -> https://github.com/dmontagu/fastapi-utils
5"""
7import inspect 1a
8from collections.abc import Callable 1a
9from typing import Any, ClassVar, ForwardRef, cast, get_origin, get_type_hints 1a
11from fastapi import APIRouter, Depends 1a
12from fastapi.routing import APIRoute 1a
13from starlette.routing import Route, WebSocketRoute 1a
15CBV_CLASS_KEY = "__cbv_class__" 1a
16INCLUDE_INIT_PARAMS_KEY = "__include_init_params__" 1a
17RETURN_TYPES_FUNC_KEY = "__return_types_func__" 1a
20def controller[T](router: APIRouter, *urls: str) -> Callable[[type[T]], type[T]]: 1a
21 """
22 This function returns a decorator that converts the decorated into a class-based view for the provided router.
23 Any methods of the decorated class that are decorated as endpoints using the router provided to this function
24 will become endpoints in the router. The first positional argument to the methods (typically `self`)
25 will be populated with an instance created using FastAPI's dependency-injection.
26 For more detail, review the documentation at
27 https://fastapi-utils.davidmontague.xyz/user-guide/class-based-views/#the-cbv-decorator
28 """
30 def decorator(cls: type[T]) -> type[T]: 1a
31 # Define cls as cbv class exclusively when using the decorator
32 return _cbv(router, cls, *urls) 1a
34 return decorator 1a
37def _cbv[T](router: APIRouter, cls: type[T], *urls: str, instance: Any | None = None) -> type[T]: 1a
38 """
39 Replaces any methods of the provided class `cls` that are endpoints of routes in `router` with updated
40 function calls that will properly inject an instance of `cls`.
41 """
42 _init_cbv(cls, instance) 1a
43 _register_endpoints(router, cls, *urls) 1a
44 return cls 1a
47# copied from Pydantic V1 Source: https://github.com/pydantic/pydantic/blob/1c91c8627b541b22354b9ed56b9ef1bb21ac6fbd/pydantic/v1/typing.py
48def _check_classvar(v: type[Any] | None) -> bool: 1a
49 if v is None: 1a
50 return False 1a
52 return v.__class__ == ClassVar.__class__ and getattr(v, "_name", None) == "ClassVar" 1a
55# copied from Pydantic V1 Source: https://github.com/pydantic/pydantic/blob/1c91c8627b541b22354b9ed56b9ef1bb21ac6fbd/pydantic/v1/typing.py
56def _is_classvar(ann_type: type[Any]) -> bool: 1a
57 if _check_classvar(ann_type) or _check_classvar(get_origin(ann_type)): 57 ↛ 58line 57 didn't jump to line 58 because the condition on line 57 was never true1a
58 return True
60 if ann_type.__class__ == ForwardRef and ann_type.__forward_arg__.startswith("ClassVar["): # type: ignore 60 ↛ 61line 60 didn't jump to line 61 because the condition on line 60 was never true1a
61 return True
63 return False 1a
66def _init_cbv(cls: type[Any], instance: Any | None = None) -> None: 1a
67 """
68 Idempotently modifies the provided `cls`, performing the following modifications:
69 * The `__init__` function is updated to set any class-annotated dependencies as instance attributes
70 * The `__signature__` attribute is updated to indicate to FastAPI what arguments should be passed to the initializer
71 * Variables starting with `_` are NOT included in the `__signature__` and will be set to None.
72 """
73 if getattr(cls, CBV_CLASS_KEY, False): # pragma: no cover 73 ↛ 74line 73 didn't jump to line 74 because the condition on line 73 was never true1a
74 return # Already initialized
75 old_init: Callable[..., Any] = cls.__init__ 1a
76 old_signature = inspect.signature(old_init) 1a
77 old_parameters = list(old_signature.parameters.values())[1:] # drop `self` parameter 1a
78 new_parameters = [ 1a
79 x for x in old_parameters if x.kind not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
80 ]
82 private_attributes = [] 1a
84 dependency_names: list[str] = [] 1a
85 for name, hint in get_type_hints(cls).items(): 1a
86 if _is_classvar(hint): 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true1a
87 continue
89 if name.startswith("_"): 1a
90 private_attributes.append(name) 1a
91 continue 1a
93 parameter_kwargs = {"default": getattr(cls, name, Ellipsis)} 1a
94 dependency_names.append(name) 1a
95 new_parameters.append( 1a
96 inspect.Parameter(name=name, kind=inspect.Parameter.KEYWORD_ONLY, annotation=hint, **parameter_kwargs)
97 )
98 new_signature = inspect.Signature(()) 1a
99 if not instance or hasattr(cls, INCLUDE_INIT_PARAMS_KEY): 99 ↛ 102line 99 didn't jump to line 102 because the condition on line 99 was always true1a
100 new_signature = old_signature.replace(parameters=new_parameters) 1a
102 def new_init(self: Any, *args: Any, **kwargs: Any) -> None: 1a
103 for dep_name in dependency_names: 1bcdefghijklmnopqrstuvwxyzABCD
104 dep_value = kwargs.pop(dep_name) 1bcdefghijklmnopqrstuvwxyzABCD
105 setattr(self, dep_name, dep_value) 1bcdefghijklmnopqrstuvwxyzABCD
106 if instance and not hasattr(cls, INCLUDE_INIT_PARAMS_KEY): 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true1bcdefghijklmnopqrstuvwxyzABCD
107 self.__class__ = instance.__class__
108 self.__dict__ = instance.__dict__
109 else:
110 old_init(self, *args, **kwargs) 1bcdefghijklmnopqrstuvwxyzABCD
112 cls.__signature__ = new_signature 1a
113 cls.__init__ = new_init 1a
114 setattr(cls, CBV_CLASS_KEY, True) 1a
116 for name in private_attributes: 1a
117 setattr(cls, name, None) 1a
120def _register_endpoints(router: APIRouter, cls: type[Any], *urls: str) -> None: 1a
121 cbv_router = APIRouter() 1a
122 function_members = inspect.getmembers(cls, inspect.isfunction) 1a
123 for url in urls: 123 ↛ 124line 123 didn't jump to line 124 because the loop on line 123 never started1a
124 _allocate_routes_by_method_name(router, url, function_members)
125 router_roles = [] 1a
126 for route in router.routes: 1a
127 assert isinstance(route, APIRoute) 1a
128 route_methods: Any = route.methods 1a
129 cast(tuple[Any], route_methods) 1a
130 router_roles.append((route.path, tuple(route_methods))) 1a
132 if len(set(router_roles)) != len(router_roles): 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true1a
133 raise Exception("An identical route role has been implemented more then once")
135 numbered_routes_by_endpoint = { 1a
136 route.endpoint: (i, route) for i, route in enumerate(router.routes) if isinstance(route, Route | WebSocketRoute)
137 }
139 prefix_length = len(router.prefix) 1a
140 routes_to_append: list[tuple[int, Route | WebSocketRoute]] = [] 1a
141 for _, func in function_members: 1a
142 index_route = numbered_routes_by_endpoint.get(func) 1a
144 if index_route is None: 1a
145 continue 1a
147 _, route = index_route 1a
148 route.path = route.path[prefix_length:] 1a
149 routes_to_append.append(index_route) 1a
150 router.routes.remove(route) 1a
152 _update_cbv_route_endpoint_signature(cls, route) 1a
153 routes_to_append.sort(key=lambda x: x[0]) 1a
155 cbv_router.routes = [route for _, route in routes_to_append] 1a
157 # In order to use a "" as a router and utilize the prefix in the original router
158 # we need to create an intermediate prefix variable to hold the prefix and pass it
159 # into the original router when using "include_router" after we reeset the original
160 # prefix. This limits the original routers usability to only the controller.
161 #
162 # This is sort of a hack and causes unexpected behavior. I'm unsure of a better solution.
163 cbv_prefix = router.prefix 1a
164 router.prefix = "" 1a
165 router.include_router(cbv_router, prefix=cbv_prefix) 1a
168def _allocate_routes_by_method_name(router: APIRouter, url: str, function_members: list[tuple[str, Any]]) -> None: 1a
169 # sourcery skip: merge-nested-ifs
170 existing_routes_endpoints: list[tuple[Any, str]] = [
171 (route.endpoint, route.path) for route in router.routes if isinstance(route, APIRoute)
172 ]
173 for name, func in function_members:
174 if hasattr(router, name) and not name.startswith("__") and not name.endswith("__"):
175 if (func, url) not in existing_routes_endpoints:
176 response_model = None
177 responses = None
178 kwargs = {}
179 status_code = 200
181 if return_types_func := getattr(func, RETURN_TYPES_FUNC_KEY, None):
182 response_model, status_code, responses, kwargs = return_types_func()
184 api_resource = router.api_route(
185 url,
186 methods=[name.capitalize()],
187 response_model=response_model,
188 status_code=status_code,
189 responses=responses,
190 **kwargs,
191 )
192 api_resource(func)
195def _update_cbv_route_endpoint_signature(cls: type[Any], route: Route | WebSocketRoute) -> None: 1a
196 """
197 Fixes the endpoint signature for a cbv route to ensure FastAPI performs dependency injection properly.
198 """
199 old_endpoint = route.endpoint 1a
200 old_signature = inspect.signature(old_endpoint) 1a
201 old_parameters: list[inspect.Parameter] = list(old_signature.parameters.values()) 1a
202 old_first_parameter = old_parameters[0] 1a
203 new_first_parameter = old_first_parameter.replace(default=Depends(cls)) 1a
204 new_parameters = [new_first_parameter] + [ 1a
205 parameter.replace(kind=inspect.Parameter.KEYWORD_ONLY) for parameter in old_parameters[1:]
206 ]
208 new_signature = old_signature.replace(parameters=new_parameters) 1a
209 route.endpoint.__signature__ = new_signature 1a