Coverage for src/lumen/api/metadata.py: 92%

1""" 

2This module constructs and returns the CapabilityStatement for the FHIR server. 

3It details supported resource types, interactions, security profiles, and search parameters. 

4""" 

5 

6from __future__ import annotations 

7 

8import json 

9from datetime import date 

10from typing import Annotated, Any 

11from uuid import UUID 

12 

13import asyncpg 

14from fastapi import APIRouter, Depends 

15from fastapi.responses import Response 

16 

17from ..auth.jwks import discover_oidc_config 

18from ..config import Settings, get_settings 

19from ..core.exceptions import FHIR_JSON 

20from ..db.pool import get_pool 

21from ..fhir.params import SearchParamDef 

22from ..fhir.resource_types import known_resource_types 

23from ..search.registry import get_registry 

24from .operations import operations_for 

25 

26router = APIRouter() 

27 

28_INTERACTIONS = [ 

29 {"code": "read"}, 

30 {"code": "vread"}, 

31 {"code": "history-instance"}, 

32 {"code": "search-type"}, 

33 {"code": "create"}, 

34 {"code": "update"}, 

35 {"code": "patch"}, 

36 {"code": "delete"}, 

37 {"code": "history-type"}, 

38] 

39 

40_OAUTH_URIS_EXT = "http://fhir-registry.smarthealthit.org/StructureDefinition/oauth-uris" 

41 

42 

43def _search_params(params: list[SearchParamDef]) -> list[dict[str, str]]: 

44 result: list[dict[str, str]] = [] 

45 for sp in sorted(params, key=lambda sp: sp.name): 

46 entry: dict[str, str] = {"name": sp.name, "type": sp.declared_type or sp.type} 

47 if sp.definition: 

48 entry["definition"] = sp.definition 

49 result.append(entry) 

50 return result 

51 

52 

53async def _resource_list( 

54 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

55 tenant_id: UUID, 

56) -> list[dict[str, Any]]: 

57 registry = get_registry() 

58 result: list[dict[str, Any]] = [] 

59 for t in sorted(known_resource_types()): 

60 # Use fully merged params (base R4B + package-loaded custom params for this tenant). 

61 params = await registry.get_params(conn, tenant_id, t) 

62 result.append( 

63 { 

64 "type": t, 

65 "interaction": _INTERACTIONS, 

66 "conditionalCreate": True, 

67 "conditionalUpdate": True, 

68 "conditionalDelete": "single", 

69 "searchParam": _search_params(params), 

70 "operation": operations_for(t), 

71 } 

72 ) 

73 return result 

74 

75 

76def _security_block(authorize_url: str, token_url: str) -> dict[str, Any]: 

77 return { 

78 "cors": True, 

79 "service": [ 

80 { 

81 "coding": [ 

82 { 

83 "system": "http://terminology.hl7.org/CodeSystem/restful-security-service", 

84 "code": "SMART-on-FHIR", 

85 "display": "SMART-on-FHIR", 

86 } 

87 ] 

88 } 

89 ], 

90 "extension": [ 

91 { 

92 "url": _OAUTH_URIS_EXT, 

93 "extension": [ 

94 {"url": "authorize", "valueUri": authorize_url}, 

95 {"url": "token", "valueUri": token_url}, 

96 ], 

97 } 

98 ], 

99 } 

100 

101 

102def _resource_list_base() -> list[dict[str, Any]]: 

103 """Return resource list using base R4B params only (no DB access).""" 

104 registry = get_registry() 

105 return [ 

106 { 

107 "type": t, 

108 "interaction": _INTERACTIONS, 

109 "conditionalCreate": True, 

110 "conditionalUpdate": True, 

111 "conditionalDelete": "single", 

112 "searchParam": _search_params(registry.get_base_params(t)), 

113 "operation": operations_for(t), 

114 } 

115 for t in sorted(known_resource_types()) 

116 ] 

117 

118 

119async def _assemble_capability( 

120 settings: Settings, 

121 resource_list: list[dict[str, Any]], 

122) -> dict[str, Any]: 

123 rest_entry: dict[str, Any] = { 

124 "mode": "server", 

125 # System-level interactions Lumen supports (see api/bundle.py). 

126 "interaction": [{"code": "transaction"}, {"code": "batch"}], 

127 "resource": resource_list, 

128 } 

129 

130 if settings.smart.enabled and settings.smart.tenant_map: 

131 issuer = next(iter(settings.smart.tenant_map)) 

132 fetch_base = settings.smart.internal_issuer_map.get(issuer) 

133 try: 

134 oidc = await discover_oidc_config(issuer, fetch_base=fetch_base) 

135 rest_entry["security"] = _security_block( 

136 authorize_url=oidc.get("authorization_endpoint", ""), 

137 token_url=oidc["token_endpoint"], 

138 ) 

139 except Exception: 

140 # If IdP is unreachable, serve the CapabilityStatement without security block 

141 # rather than failing the entire metadata endpoint. 

142 pass 

143 

144 return { 

145 "resourceType": "CapabilityStatement", 

146 "status": "active", 

147 "date": str(date.today()), 

148 "kind": "instance", 

149 "software": {"name": "Lumen"}, 

150 "implementation": {"description": "Lumen FHIR R4B Server"}, 

151 "fhirVersion": "4.3.0", 

152 "format": ["application/fhir+json", "application/fhir+xml", "application/json"], 

153 "rest": [rest_entry], 

154 } 

155 

156 

157async def _build_capability_base(settings: Settings) -> dict[str, Any]: 

158 """Build CapabilityStatement using base R4B params only (no DB required).""" 

159 return await _assemble_capability(settings, _resource_list_base()) 

160 

161 

162async def _build_capability( 

163 settings: Settings, 

164 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

165 tenant_id: UUID, 

166) -> dict[str, Any]: 

167 return await _assemble_capability(settings, await _resource_list(conn, tenant_id)) 

168 

169 

170@router.get("/metadata") 

171async def metadata( 

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

173) -> Response: 

174 try: 

175 pool = get_pool() 

176 async with pool.acquire() as conn: 

177 body = await _build_capability(settings, conn, settings.default_tenant) 

178 except RuntimeError: 

179 # Pool not yet initialised (e.g. during startup health-checks or tests). 

180 # Fall back to base R4B params only — custom/package params are omitted. 

181 body = await _build_capability_base(settings) 

182 return Response(content=json.dumps(body), media_type=FHIR_JSON)