Coverage for src/lumen/api/admin/derivation_rules.py: 100%

1""" 

2This router manages tenant-scoped derivation rules (write-time field 

3derivation via ConceptMap). Rules are plain admin-configuration data, not 

4FHIR resources — there is no standard FHIR resource type for "derive field X 

5from field Y through this ConceptMap". 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import Any 

11 

12from fastapi import APIRouter, HTTPException, status 

13from fastapi.responses import JSONResponse 

14from pydantic import BaseModel 

15 

16from ...db.pool import get_pool 

17from ...derivation.registry import get_registry 

18from ..deps import TenantDep 

19 

20router = APIRouter() 

21 

22 

23class DerivationRuleUpsert(BaseModel): 

24 resource_type: str 

25 target_path: str 

26 source_path: str 

27 concept_map_url: str 

28 description: str | None = None 

29 target_is_list: bool = False 

30 

31 

32def _serialize(row: Any) -> dict[str, Any]: 

33 return { 

34 "resource_type": row["resource_type"], 

35 "target_path": row["target_path"], 

36 "source_path": row["source_path"], 

37 "concept_map_url": row["concept_map_url"], 

38 "description": row["description"], 

39 "target_is_list": row["target_is_list"], 

40 } 

41 

42 

43@router.post("/derivation-rules", status_code=status.HTTP_201_CREATED) 

44async def upsert_derivation_rule( 

45 tenant_id: TenantDep, 

46 body: DerivationRuleUpsert, 

47) -> JSONResponse: 

48 """Register (or replace) a derivation rule for a resource type + target field.""" 

49 pool = get_pool() 

50 async with pool.acquire() as conn: 

51 row = await conn.fetchrow( 

52 """ 

53 INSERT INTO derivation_rules 

54 (tenant_id, resource_type, target_path, source_path, concept_map_url, 

55 description, target_is_list) 

56 VALUES ($1, $2, $3, $4, $5, $6, $7) 

57 ON CONFLICT (tenant_id, resource_type, target_path) 

58 DO UPDATE SET 

59 source_path = EXCLUDED.source_path, 

60 concept_map_url = EXCLUDED.concept_map_url, 

61 description = EXCLUDED.description, 

62 target_is_list = EXCLUDED.target_is_list, 

63 updated_at = now() 

64 RETURNING resource_type, target_path, source_path, concept_map_url, 

65 description, target_is_list 

66 """, 

67 tenant_id, 

68 body.resource_type, 

69 body.target_path, 

70 body.source_path, 

71 body.concept_map_url, 

72 body.description, 

73 body.target_is_list, 

74 ) 

75 get_registry().invalidate(tenant_id) 

76 return JSONResponse(content=_serialize(row), status_code=status.HTTP_201_CREATED) 

77 

78 

79@router.get("/derivation-rules") 

80async def list_derivation_rules(tenant_id: TenantDep) -> JSONResponse: 

81 """List all derivation rules registered for the current tenant.""" 

82 pool = get_pool() 

83 async with pool.acquire() as conn: 

84 rows = await conn.fetch( 

85 """ 

86 SELECT resource_type, target_path, source_path, concept_map_url, 

87 description, target_is_list 

88 FROM derivation_rules 

89 WHERE tenant_id = $1 

90 ORDER BY resource_type, target_path 

91 """, 

92 tenant_id, 

93 ) 

94 return JSONResponse(content={"rules": [_serialize(row) for row in rows]}) 

95 

96 

97@router.delete( 

98 "/derivation-rules/{resource_type}/{target_path}", status_code=status.HTTP_204_NO_CONTENT 

99) 

100async def delete_derivation_rule( 

101 tenant_id: TenantDep, 

102 resource_type: str, 

103 target_path: str, 

104) -> None: 

105 """Remove a derivation rule.""" 

106 pool = get_pool() 

107 async with pool.acquire() as conn: 

108 result = await conn.execute( 

109 """ 

110 DELETE FROM derivation_rules 

111 WHERE tenant_id = $1 AND resource_type = $2 AND target_path = $3 

112 """, 

113 tenant_id, 

114 resource_type, 

115 target_path, 

116 ) 

117 get_registry().invalidate(tenant_id) 

118 if result == "DELETE 0": 

119 raise HTTPException( 

120 status_code=status.HTTP_404_NOT_FOUND, 

121 detail=f"No derivation rule for {resource_type}.{target_path}", 

122 )