Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/common_utils/get_routes.py: 87%

1""" 

2Utility class for getting routes from a FastAPI app. 

3""" 

4 

5from collections.abc import Sequence 

6from typing import Final, Protocol 

7 

8from starlette.routing import BaseRoute 

9from typing_extensions import NotRequired, ReadOnly, TypedDict 

10 

11from litellm._logging import verbose_logger 

12 

13 

14class NamedEndpoint(Protocol): 

15 __name__: str 

16 

17 

18class RouteInfo(TypedDict): 

19 path: ReadOnly[str | None] 

20 methods: ReadOnly[Sequence[str] | None] 

21 name: ReadOnly[str | None] 

22 endpoint: ReadOnly[str | None] 

23 mounted_app: NotRequired[ReadOnly[bool]] 

24 

25 

26class GetRoutes: 

27 @staticmethod 

28 def get_app_routes( 

29 route: BaseRoute, 

30 endpoint_route: NamedEndpoint, 

31 ) -> list[RouteInfo]: 

32 """ 

33 Get routes for a regular route. 

34 """ 

35 route_info: Final[RouteInfo] = { 

36 "path": getattr(route, "path", None), 

37 "methods": getattr(route, "methods", None), 

38 "name": getattr(route, "name", None), 

39 "endpoint": (endpoint_route.__name__ if getattr(route, "endpoint", None) else None), 

40 } 

41 return [route_info] 

42 

43 @staticmethod 

44 def get_routes_for_mounted_app( 

45 route: BaseRoute, 

46 ) -> list[RouteInfo]: 

47 """ 

48 Get routes for a mounted sub-application. 

49 """ 

50 mount_path: Final[str] = getattr(route, "path", "") 

51 sub_app: Final[object] = getattr(route, "app", None) 

52 if not sub_app or not hasattr(sub_app, "routes"): 

53 return [] 

54 sub_routes: Final[Sequence[object]] = getattr(sub_app, "routes", ()) 

55 return [ 

56 sub_route_info 

57 for sub_route in sub_routes 

58 if (sub_route_info := GetRoutes._mounted_sub_route_info(mount_path, sub_route)) is not None 

59 ] 

60 

61 @staticmethod 

62 def _mounted_sub_route_info(mount_path: str, sub_route: object) -> RouteInfo | None: 

63 endpoint_func: Final[object] = getattr(sub_route, "endpoint", None) or getattr(sub_route, "app", None) 

64 if endpoint_func is None: 64 ↛ 65line 64 didn't jump to line 65 because the condition on line 64 was never true

65 return None 

66 sub_route_path: Final[str] = getattr(sub_route, "path", "") 

67 return { 

68 "path": mount_path.rstrip("/") + sub_route_path, 

69 "methods": getattr(sub_route, "methods", ["GET", "POST"]), 

70 "name": getattr(sub_route, "name", None), 

71 "endpoint": GetRoutes._safe_get_endpoint_name(endpoint_func), 

72 "mounted_app": True, 

73 } 

74 

75 @staticmethod 

76 def _safe_get_endpoint_name(endpoint_function: object) -> str | None: 

77 """ 

78 Safely get the name of the endpoint function. 

79 """ 

80 try: 

81 if hasattr(endpoint_function, "__name__"): 

82 endpoint_name: Final[str] = getattr(endpoint_function, "__name__", "") 

83 return endpoint_name 

84 elif hasattr(endpoint_function, "__class__") and hasattr(endpoint_function.__class__, "__name__"): 84 ↛ 87line 84 didn't jump to line 87 because the condition on line 84 was always true

85 return endpoint_function.__class__.__name__ 

86 else: 

87 return None 

88 except Exception: 

89 verbose_logger.exception("Error getting endpoint name for route: %s", endpoint_function) 

90 return None