Coverage for src/lumen/api/tenant_settings.py: 90%

1""" 

2This router defines routes for managing tenant-specific configuration settings. 

3It is primarily used by the admin UI to adjust server behavior per tenant. 

4""" 

5 

6from __future__ import annotations 

7 

8import json 

9from typing import Any, Literal 

10 

11import structlog 

12from fastapi import APIRouter, Response 

13from fastapi.responses import JSONResponse 

14from pydantic import BaseModel 

15 

16from ..audit.events import build_audit_event 

17from ..auth.enforcement import ScopeEnforcerDep 

18from ..config import get_settings 

19from ..db.pool import get_pool 

20from ..db.repository import ResourceRepository 

21from ..search.registry import get_registry 

22from .deps import TenantSettingsDep 

23 

24logger = structlog.get_logger(__name__) 

25 

26router = APIRouter(dependencies=[ScopeEnforcerDep]) 

27 

28 

29class TenantSettingsUpdate(BaseModel): 

30 create_on_update: bool | None = None 

31 reference_integrity: Literal["strict", "lenient", "disabled"] | None = None 

32 validation_strictness: Literal["strict", "lenient"] | None = None 

33 terminology_strictness: Literal["strict", "lenient"] | None = None 

34 timezone_strictness: Literal["strict", "lenient", "disabled"] | None = None 

35 profile_validation_mode: Literal["tolerant", "declared", "enforced"] | None = None 

36 enforced_profiles: dict[str, list[str]] | None = None 

37 terminology_pre_expansion_enabled: bool | None = None 

38 ignore_unknown_search_parameters: bool | None = None 

39 

40 

41@router.get("/tenant_settings") 

42async def get_settings_endpoint( 

43 tenant_settings: TenantSettingsDep, 

44) -> Response: 

45 resolved = { 

46 "create_on_update": tenant_settings.create_on_update, 

47 "reference_integrity": tenant_settings.reference_integrity, 

48 "validation_strictness": tenant_settings.validation_strictness, 

49 "terminology_strictness": tenant_settings.terminology_strictness, 

50 "timezone_strictness": tenant_settings.timezone_strictness, 

51 "profile_validation_mode": tenant_settings.profile_validation_mode, 

52 "enforced_profiles": tenant_settings.enforced_profiles, 

53 "terminology_pre_expansion_enabled": tenant_settings.terminology_pre_expansion_enabled, 

54 "ignore_unknown_search_parameters": tenant_settings.ignore_unknown_search_parameters, 

55 } 

56 return JSONResponse(content=resolved) 

57 

58 

59@router.put("/tenant_settings") 

60async def update_settings_endpoint( 

61 body: TenantSettingsUpdate, 

62 tenant_settings: TenantSettingsDep, 

63) -> Response: 

64 new_settings: dict[str, Any] = {} 

65 if body.create_on_update is not None: 

66 new_settings["create_on_update"] = body.create_on_update 

67 if body.reference_integrity is not None: 

68 new_settings["reference_integrity"] = body.reference_integrity 

69 if body.validation_strictness is not None: 

70 new_settings["validation_strictness"] = body.validation_strictness 

71 if body.terminology_strictness is not None: 

72 new_settings["terminology_strictness"] = body.terminology_strictness 

73 if body.timezone_strictness is not None: 

74 new_settings["timezone_strictness"] = body.timezone_strictness 

75 if body.profile_validation_mode is not None: 

76 new_settings["profile_validation_mode"] = body.profile_validation_mode 

77 if body.enforced_profiles is not None: 

78 new_settings["enforced_profiles"] = body.enforced_profiles 

79 if body.terminology_pre_expansion_enabled is not None: 

80 new_settings["terminology_pre_expansion_enabled"] = body.terminology_pre_expansion_enabled 

81 if body.ignore_unknown_search_parameters is not None: 

82 new_settings["ignore_unknown_search_parameters"] = body.ignore_unknown_search_parameters 

83 

84 pool = get_pool() 

85 async with pool.acquire() as conn: 

86 if new_settings: 

87 await conn.execute( 

88 """ 

89 INSERT INTO tenant_settings (tenant_id, settings) 

90 VALUES ($1, $2::jsonb) 

91 ON CONFLICT (tenant_id) DO UPDATE 

92 SET settings = EXCLUDED.settings 

93 """, 

94 tenant_settings.tenant_id, 

95 json.dumps(new_settings), 

96 ) 

97 logger.info("tenant_settings.updated", new_settings=new_settings) 

98 

99 # Generate AuditEvent 

100 from uuid import uuid4 

101 

102 audit_id = uuid4().hex 

103 audit_event = build_audit_event("U", "TenantSettings", "global") 

104 audit_event["id"] = audit_id 

105 await ResourceRepository(conn, get_registry()).create( 

106 tenant_settings.tenant_id, "AuditEvent", audit_id, audit_event 

107 ) 

108 else: 

109 await conn.execute( 

110 "DELETE FROM tenant_settings WHERE tenant_id = $1", 

111 tenant_settings.tenant_id, 

112 ) 

113 logger.info("tenant_settings.cleared") 

114 

115 # Generate AuditEvent 

116 from uuid import uuid4 

117 

118 audit_id = uuid4().hex 

119 audit_event = build_audit_event("D", "TenantSettings", "global") 

120 audit_event["id"] = audit_id 

121 await ResourceRepository(conn, get_registry()).create( 

122 tenant_settings.tenant_id, "AuditEvent", audit_id, audit_event 

123 ) 

124 

125 global_settings = get_settings() 

126 resolved = { 

127 "create_on_update": new_settings.get( 

128 "create_on_update", global_settings.tenant.create_on_update 

129 ), 

130 "reference_integrity": new_settings.get( 

131 "reference_integrity", global_settings.tenant.reference_integrity 

132 ), 

133 "validation_strictness": new_settings.get( 

134 "validation_strictness", global_settings.tenant.validation_strictness 

135 ), 

136 "terminology_strictness": new_settings.get( 

137 "terminology_strictness", global_settings.tenant.terminology_strictness 

138 ), 

139 "timezone_strictness": new_settings.get( 

140 "timezone_strictness", global_settings.tenant.timezone_strictness 

141 ), 

142 "profile_validation_mode": new_settings.get( 

143 "profile_validation_mode", global_settings.tenant.profile_validation_mode 

144 ), 

145 "enforced_profiles": new_settings.get( 

146 "enforced_profiles", global_settings.tenant.enforced_profiles 

147 ), 

148 "terminology_pre_expansion_enabled": new_settings.get( 

149 "terminology_pre_expansion_enabled", 

150 global_settings.tenant.terminology_pre_expansion_enabled, 

151 ), 

152 "ignore_unknown_search_parameters": new_settings.get( 

153 "ignore_unknown_search_parameters", 

154 global_settings.tenant.ignore_unknown_search_parameters, 

155 ), 

156 } 

157 return JSONResponse(content=resolved)