Coverage for src/lumen/cortex/context.py: 86%

1""" 

2The two ports a plugin gets: ``ctx.read`` and ``ctx.write``. 

3 

4Both are *handles*, not access paths. They exist only inside a plugin running 

5in the server process; nothing here opens a second way into the data for an 

6outside caller. 

7 

8What the write port guarantees, in one place: 

9 

10* **The same validation as HTTP.** Every write runs the full inbound stack — 

11 JSON Schema, base code bindings, terminology bindings, profiles — under the 

12 writing tenant's settings, exactly as ``POST``/``PUT`` would. 

13* **The same write semantics.** Conditional create (``if_none_exist``), 

14 conditional update, create-on-update, internal ``urn:uuid`` references and 

15 atomic transaction bundles are all available, because it is the same engine. 

16* **The same audit.** ``AuditEvent`` is written in the repository, below both 

17 transports. 

18* **Attribution on top.** Everything written here is tagged as plugin-written; 

19 this is the one thing the in-process path adds that HTTP does not have, and 

20 it is added above the shared layer, never inside it. 

21 

22What it does *not* do: SMART scope enforcement. Scopes are checked on the HTTP 

23edge against a bearer token, and a plugin has no token. A plugin therefore 

24writes with the full rights of its tenant, which is why plugins are curated 

25deployment configuration and not user input. 

26""" 

27 

28from __future__ import annotations 

29 

30from dataclasses import dataclass, field 

31from typing import Any 

32from uuid import UUID 

33 

34import structlog 

35 

36from ..api.deps import ResolvedTenantSettings 

37from ..core.exceptions import OperationOutcomeError 

38from ..service import ResourceService, WriteOutcome 

39from .provenance import AiAttribution, apply_attribution, build_provenance 

40 

41logger = structlog.get_logger(__name__) 

42 

43SearchValue = str | int | float | list[str] 

44 

45 

46class ReadPort: 

47 """Context lookups without HTTP: does it exist, what else is there.""" 

48 

49 def __init__(self, service: ResourceService) -> None: 

50 self._service = service 

51 

52 async def read(self, resource_type: str, resource_id: str) -> dict[str, Any] | None: 

53 """The current version of one resource, or None if absent or deleted.""" 

54 result = await self._service.read(resource_type, resource_id) 

55 return result["resource"] if result else None 

56 

57 async def exists(self, resource_type: str, resource_id: str) -> bool: 

58 return await self._service.read(resource_type, resource_id) is not None 

59 

60 async def search(self, resource_type: str, **params: SearchValue) -> list[dict[str, Any]]: 

61 """Search as the REST API would, e.g. ``search("Observation", code="8310-5")``. 

62 

63 Parameter names are the FHIR ones. Since a Python keyword cannot 

64 contain ``-``, write it as ``__``: ``value__quantity`` becomes 

65 ``value-quantity``. Names starting with ``_`` (``_count``, 

66 ``_lastUpdated``) pass through unchanged. 

67 """ 

68 return await self._service.search(resource_type, _as_query(params)) 

69 

70 async def match(self, resource_type: str, query: str) -> list[str]: 

71 """Resolve a conditional query string to the ids it matches.""" 

72 return await self._service.conditional_search_ids(resource_type, query) 

73 

74 

75def _as_query(params: dict[str, SearchValue]) -> dict[str, list[str]]: 

76 query: dict[str, list[str]] = {} 

77 for key, value in params.items(): 

78 name = key if key.startswith("_") else key.replace("__", "-") 

79 values = value if isinstance(value, list) else [str(value)] 

80 query.setdefault(name, []).extend(str(v) for v in values) 

81 return query 

82 

83 

84class WritePort: 

85 """The full write semantics of the HTTP API, as function calls.""" 

86 

87 def __init__(self, service: ResourceService, plugin_name: str, plugin_version: str) -> None: 

88 self._service = service 

89 self._plugin = plugin_name 

90 self._version = plugin_version 

91 

92 async def create( 

93 self, 

94 resource: dict[str, Any], 

95 *, 

96 if_none_exist: str | None = None, 

97 ai: AiAttribution | None = None, 

98 ) -> dict[str, Any]: 

99 """Create a resource. Returns it as stored, with ``meta`` filled in.""" 

100 resource_type = _resource_type(resource) 

101 prepared = apply_attribution(resource, self._plugin, ai) 

102 validated = await self._service.validate_resource_dict(prepared, resource_type) 

103 outcome = await self._service.create(resource_type, validated, if_none_exist=if_none_exist) 

104 await self._record_provenance(resource_type, outcome, ai) 

105 return outcome.stored_resource 

106 

107 async def update( 

108 self, 

109 resource: dict[str, Any], 

110 *, 

111 ai: AiAttribution | None = None, 

112 ) -> dict[str, Any]: 

113 """Update the resource identified by its own ``id``.""" 

114 resource_type = _resource_type(resource) 

115 resource_id = resource.get("id") 

116 if not isinstance(resource_id, str) or not resource_id: 

117 raise OperationOutcomeError(400, "invalid", "update() needs a resource with an id") 

118 prepared = apply_attribution(resource, self._plugin, ai) 

119 validated = await self._service.validate_resource_dict(prepared, resource_type) 

120 outcome = await self._service.update(resource_type, resource_id, validated) 

121 await self._record_provenance(resource_type, outcome, ai) 

122 return outcome.stored_resource 

123 

124 async def conditional_update( 

125 self, 

126 resource: dict[str, Any], 

127 query: str, 

128 *, 

129 ai: AiAttribution | None = None, 

130 ) -> dict[str, Any]: 

131 """``PUT /{type}?{query}`` — update the single match, or create it.""" 

132 resource_type = _resource_type(resource) 

133 prepared = apply_attribution(resource, self._plugin, ai) 

134 validated = await self._service.validate_resource_dict(prepared, resource_type) 

135 outcome = await self._service.conditional_update(resource_type, query, validated) 

136 await self._record_provenance(resource_type, outcome, ai) 

137 return outcome.stored_resource 

138 

139 async def delete(self, resource_type: str, resource_id: str) -> bool: 

140 """Delete a resource. False means it never existed.""" 

141 return await self._service.delete(resource_type, resource_id) 

142 

143 async def transaction(self, bundle: dict[str, Any]) -> dict[str, Any]: 

144 """Submit a transaction or batch Bundle — atomicity and ``urn:uuid`` included. 

145 

146 This is the same code path as ``POST /`` over REST, so internal 

147 references between entries resolve exactly as they do there. Entry 

148 resources are *not* tagged: a bundle is a container, and rewriting the 

149 resources inside it would change what the plugin asked to store. 

150 """ 

151 from ..api.bundle import process_bundle 

152 

153 return await process_bundle( 

154 bundle, 

155 self._service.pool, 

156 self._service.tenant_id, 

157 self._service.tenant_settings, 

158 _base_url(), 

159 ) 

160 

161 async def _record_provenance( 

162 self, 

163 resource_type: str, 

164 outcome: WriteOutcome, 

165 ai: AiAttribution | None, 

166 ) -> None: 

167 """Write the Provenance for an AI-attributed write, if one was asked for.""" 

168 if ai is None or not ai.provenance: 

169 return 

170 provenance = build_provenance( 

171 target_reference=f"{resource_type}/{outcome.resource_id}", 

172 plugin_name=self._plugin, 

173 plugin_version=self._version, 

174 ai=ai, 

175 ) 

176 try: 

177 validated = await self._service.validate_resource_dict(provenance, "Provenance") 

178 await self._service.create("Provenance", validated) 

179 except OperationOutcomeError as exc: 

180 # The attributed resource is already stored and tagged; losing its 

181 # Provenance is worth a loud log, not an undo of clinical content. 

182 await logger.aerror( 

183 "cortex.provenance_failed", 

184 plugin=self._plugin, 

185 resource_type=resource_type, 

186 code=exc.code, 

187 diagnostics=exc.diagnostics, 

188 ) 

189 

190 

191def _resource_type(resource: dict[str, Any]) -> str: 

192 resource_type = resource.get("resourceType") 

193 if not isinstance(resource_type, str) or not resource_type: 

194 raise OperationOutcomeError(400, "invalid", "Resource has no resourceType") 

195 return resource_type 

196 

197 

198def _base_url() -> str: 

199 from ..config import get_settings 

200 

201 return get_settings().server_address.rstrip("/") 

202 

203 

204@dataclass 

205class PluginContext: 

206 """What a hook or operation is handed: who it is, and the two ports.""" 

207 

208 tenant_id: UUID 

209 plugin_name: str 

210 read: ReadPort 

211 write: WritePort 

212 #: Structured logger already bound to the plugin name. 

213 log: Any = field(default=None) 

214 

215 def __post_init__(self) -> None: 

216 if self.log is None: 

217 self.log = logger.bind(plugin=self.plugin_name) 

218 

219 

220def build_context( 

221 tenant_id: UUID, 

222 tenant_settings: ResolvedTenantSettings, 

223 plugin_name: str, 

224 plugin_version: str, 

225) -> PluginContext: 

226 """Assemble a context over a fresh service for one tenant.""" 

227 service = ResourceService(tenant_id, tenant_settings) 

228 return PluginContext( 

229 tenant_id=tenant_id, 

230 plugin_name=plugin_name, 

231 read=ReadPort(service), 

232 write=WritePort(service, plugin_name, plugin_version), 

233 )