Coverage for src/lumen/cortex/operation.py: 68%

1""" 

2What a plugin-provided custom operation is handed. 

3 

4A plugin operation is indistinguishable from a built-in one to the caller — 

5same URL shape, same ``OperationOutcome`` on error, same entry in 

6``/metadata``. What it does *not* get is the raw ``Request``: the operation 

7sees the invocation, not the transport, so the same handler stays callable 

8when it is invoked from somewhere other than HTTP. 

9""" 

10 

11from __future__ import annotations 

12 

13import json 

14from dataclasses import dataclass 

15from typing import Any 

16from uuid import UUID 

17 

18from ..core.exceptions import OperationOutcomeError 

19 

20 

21@dataclass(frozen=True) 

22class OperationInvocation: 

23 """One call to ``$name``.""" 

24 

25 name: str 

26 scope: str 

27 resource_type: str | None 

28 resource_id: str | None 

29 tenant_id: UUID 

30 #: Raw request body (empty for GET invocations). 

31 body: bytes 

32 #: Query parameters, as sent. 

33 query: dict[str, list[str]] 

34 

35 def json(self) -> dict[str, Any]: 

36 """The body parsed as a JSON object. Raises 400 if it is not one.""" 

37 if not self.body: 

38 return {} 

39 try: 

40 parsed: Any = json.loads(self.body) 

41 except json.JSONDecodeError as exc: 

42 raise OperationOutcomeError(400, "invalid", f"Invalid JSON: {exc}") from exc 

43 if not isinstance(parsed, dict): 

44 raise OperationOutcomeError(400, "invalid", "Body must be a JSON object") 

45 return parsed 

46 

47 def param(self, name: str, default: str | None = None) -> str | None: 

48 """A single query parameter, or ``default``.""" 

49 values = self.query.get(name) 

50 return values[0] if values else default