Coverage for src/lumen/api/terminology_ops.py: 97%

1""" 

2This module implements API handlers for standard FHIR terminology operations. 

3It routes requests for $lookup, $expand, and $translate to the internal terminology engine. 

4""" 

5 

6from __future__ import annotations 

7 

8import json 

9from datetime import UTC, datetime 

10from typing import Any 

11 

12import asyncpg 

13from fastapi.responses import Response 

14 

15from ..core.exceptions import FHIR_JSON, OperationOutcomeError 

16from ..db.pool import get_pool 

17from ..db.repository import ReadResult, ResourceRepository 

18from ..fhir import terminology as tx 

19from ..search.registry import get_registry 

20from .operations import OperationContext, OpScope, operation 

21 

22 

23def _parameters(parameters: list[dict[str, Any]]) -> Response: 

24 body = {"resourceType": "Parameters", "parameter": parameters} 

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

26 

27 

28def _require(ctx: OperationContext, name: str) -> str: 

29 value = ctx.request.query_params.get(name) 

30 if not value: 

31 raise OperationOutcomeError(400, "invalid", f"Missing required parameter '{name}'") 

32 return value 

33 

34 

35async def _resolve( 

36 repo: ResourceRepository, ctx: OperationContext, expected_type: str 

37) -> ReadResult: 

38 """Resolve the operation's target resource by instance id or by ``url`` param.""" 

39 if ctx.resource_type != expected_type: 

40 raise OperationOutcomeError( 

41 400, "not-supported", f"Operation is only supported on {expected_type}" 

42 ) 

43 if ctx.scope is OpScope.INSTANCE: 

44 assert ctx.resource_id is not None 

45 result = await repo.read(ctx.tenant_id, expected_type, ctx.resource_id) 

46 ref = f"{expected_type}/{ctx.resource_id}" 

47 else: 

48 url = _require(ctx, "url") 

49 result = await repo.find_current_by_url(ctx.tenant_id, expected_type, url) 

50 ref = url 

51 if result is None: 

52 raise OperationOutcomeError(404, "not-found", f"{expected_type} {ref} not found") 

53 return result 

54 

55 

56def _terminology_error(exc: tx.TerminologyError) -> OperationOutcomeError: 

57 status = 404 if exc.code == "not-found" else 400 

58 return OperationOutcomeError(status, exc.code, str(exc)) 

59 

60 

61# ── CodeSystem/$lookup ───────────────────────────────────────────────────────── 

62 

63 

64@operation("lookup", OpScope.TYPE) 

65async def op_lookup(ctx: OperationContext) -> Response: 

66 if ctx.resource_type != "CodeSystem": 

67 raise OperationOutcomeError(400, "not-supported", "$lookup is only supported on CodeSystem") 

68 system = _require(ctx, "system") 

69 code = _require(ctx, "code") 

70 

71 pool: asyncpg.Pool[asyncpg.Record] = get_pool() 

72 async with pool.acquire() as conn: 

73 cs = await ResourceRepository(conn, get_registry()).find_current_by_url( 

74 ctx.tenant_id, "CodeSystem", system 

75 ) 

76 if cs is None: 

77 raise OperationOutcomeError(404, "not-found", f"CodeSystem {system} not found") 

78 

79 concept = tx.lookup_concept(cs["resource"], code) 

80 if concept is None: 

81 raise OperationOutcomeError(404, "code-invalid", f"Code '{code}' not found in {system}") 

82 

83 params: list[dict[str, Any]] = [] 

84 if isinstance(cs["resource"].get("name"), str): 

85 params.append({"name": "name", "valueString": cs["resource"]["name"]}) 

86 if isinstance(concept.get("display"), str): 

87 params.append({"name": "display", "valueString": concept["display"]}) 

88 params.append({"name": "code", "valueCode": code}) 

89 return _parameters(params) 

90 

91 

92# ── ValueSet/$expand ─────────────────────────────────────────────────────────── 

93 

94 

95@operation("expand", OpScope.TYPE, OpScope.INSTANCE) 

96async def op_expand(ctx: OperationContext) -> Response: 

97 pool: asyncpg.Pool[asyncpg.Record] = get_pool() 

98 async with pool.acquire() as conn: 

99 repo = ResourceRepository(conn, get_registry()) 

100 value_set = (await _resolve(repo, ctx, "ValueSet"))["resource"] 

101 

102 systems: dict[str, dict[str, Any]] = {} 

103 for system_url in tx.referenced_systems(value_set): 

104 cs = await repo.find_current_by_url(ctx.tenant_id, "CodeSystem", system_url) 

105 if cs is not None: 

106 systems[system_url] = cs["resource"] 

107 

108 try: 

109 contains = tx.expand_valueset(value_set, systems) 

110 except tx.TerminologyError as exc: 

111 raise _terminology_error(exc) from exc 

112 

113 expanded = { 

114 **value_set, 

115 "expansion": { 

116 "timestamp": datetime.now(UTC).isoformat(), 

117 "total": len(contains), 

118 "contains": contains, 

119 }, 

120 } 

121 return Response(content=json.dumps(expanded), media_type=FHIR_JSON) 

122 

123 

124# ── ConceptMap/$translate ────────────────────────────────────────────────────── 

125 

126 

127@operation("translate", OpScope.TYPE, OpScope.INSTANCE) 

128async def op_translate(ctx: OperationContext) -> Response: 

129 code = _require(ctx, "code") 

130 system = ctx.request.query_params.get("system") 

131 target = ctx.request.query_params.get("target") 

132 

133 pool: asyncpg.Pool[asyncpg.Record] = get_pool() 

134 async with pool.acquire() as conn: 

135 repo = ResourceRepository(conn, get_registry()) 

136 concept_map = (await _resolve(repo, ctx, "ConceptMap"))["resource"] 

137 

138 matches = tx.translate(concept_map, system, code, target) 

139 

140 params: list[dict[str, Any]] = [{"name": "result", "valueBoolean": bool(matches)}] 

141 for match in matches: 

142 params.append( 

143 { 

144 "name": "match", 

145 "part": [ 

146 {"name": "equivalence", "valueCode": match["equivalence"]}, 

147 {"name": "concept", "valueCoding": match["concept"]}, 

148 ], 

149 } 

150 ) 

151 return _parameters(params)