Coverage for src/lumen/api/deps.py: 100%

1""" 

2This module defines shared FastAPI dependencies used across routes. 

3It provides helpers for tenant resolution, token validation, and database connection retrieval. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import Annotated, Any 

9from uuid import UUID 

10 

11from fastapi import Depends, Request 

12 

13from ..auth.token import validate_token 

14from ..config import Settings, get_settings 

15from ..db.pool import get_pool 

16 

17 

18class ResolvedTenantSettings: 

19 def __init__(self, tenant_id: UUID, db_settings: dict[str, Any], global_settings: Settings): 

20 self.tenant_id = tenant_id 

21 self._db_settings = db_settings 

22 self._global_settings = global_settings 

23 

24 @property 

25 def create_on_update(self) -> bool: 

26 if "create_on_update" in self._db_settings: 

27 return bool(self._db_settings["create_on_update"]) 

28 return self._global_settings.tenant.create_on_update 

29 

30 @property 

31 def reference_integrity(self) -> str: 

32 if "reference_integrity" in self._db_settings: 

33 return str(self._db_settings["reference_integrity"]) 

34 return self._global_settings.tenant.reference_integrity 

35 

36 @property 

37 def validation_strictness(self) -> str: 

38 if "validation_strictness" in self._db_settings: 

39 return str(self._db_settings["validation_strictness"]) 

40 return self._global_settings.tenant.validation_strictness 

41 

42 @property 

43 def terminology_strictness(self) -> str: 

44 if "terminology_strictness" in self._db_settings: 

45 return str(self._db_settings["terminology_strictness"]) 

46 return self._global_settings.tenant.terminology_strictness 

47 

48 @property 

49 def timezone_strictness(self) -> str: 

50 if "timezone_strictness" in self._db_settings: 

51 return str(self._db_settings["timezone_strictness"]) 

52 return self._global_settings.tenant.timezone_strictness 

53 

54 @property 

55 def ignore_unknown_search_parameters(self) -> bool: 

56 if "ignore_unknown_search_parameters" in self._db_settings: 

57 return bool(self._db_settings["ignore_unknown_search_parameters"]) 

58 return self._global_settings.tenant.ignore_unknown_search_parameters 

59 

60 @property 

61 def profile_validation_mode(self) -> str: 

62 if "profile_validation_mode" in self._db_settings: 

63 return str(self._db_settings["profile_validation_mode"]) 

64 return self._global_settings.tenant.profile_validation_mode 

65 

66 @property 

67 def terminology_pre_expansion_enabled(self) -> bool: 

68 if "terminology_pre_expansion_enabled" in self._db_settings: 

69 return bool(self._db_settings["terminology_pre_expansion_enabled"]) 

70 return self._global_settings.tenant.terminology_pre_expansion_enabled 

71 

72 @property 

73 def enforced_profiles(self) -> dict[str, list[str]]: 

74 if "enforced_profiles" in self._db_settings: 

75 return dict(self._db_settings["enforced_profiles"]) 

76 return self._global_settings.tenant.enforced_profiles 

77 

78 

79async def current_tenant( 

80 request: Request, 

81 settings: Annotated[Settings, Depends(get_settings)], 

82) -> UUID: 

83 if not settings.smart.enabled: 

84 return settings.default_tenant 

85 return await validate_token(request, settings.smart) 

86 

87 

88async def resolve_tenant_settings( 

89 tenant: UUID, 

90 settings: Settings | None = None, 

91) -> ResolvedTenantSettings: 

92 """Load a tenant's effective settings. Usable outside a request (Cortex).""" 

93 if settings is None: 

94 settings = get_settings() 

95 pool = get_pool() 

96 db_settings: dict[str, Any] = {} 

97 async with pool.acquire() as conn: 

98 row = await conn.fetchrow( 

99 "SELECT settings FROM tenant_settings WHERE tenant_id = $1", 

100 tenant, 

101 ) 

102 if row: 

103 import json 

104 

105 raw = row["settings"] 

106 db_settings = json.loads(raw) if isinstance(raw, str) else dict(raw) 

107 return ResolvedTenantSettings(tenant, db_settings, settings) 

108 

109 

110async def get_tenant_settings( 

111 tenant: Annotated[UUID, Depends(current_tenant)], 

112 settings: Annotated[Settings, Depends(get_settings)], 

113) -> ResolvedTenantSettings: 

114 return await resolve_tenant_settings(tenant, settings) 

115 

116 

117TenantDep = Annotated[UUID, Depends(current_tenant)] 

118TenantSettingsDep = Annotated[ResolvedTenantSettings, Depends(get_tenant_settings)]