Coverage for src/lumen/cortex/internal_plugins/analytics/plugin.py: 0%

1""" 

2AnalyticsPlugin — Built-in system plugin for high-performance analytics. 

3 

4Exposes custom FHIR operations over HTTP: 

5- Observation/$search-pandas: Returns a flat JSON list of normalised records. 

6- Observation/$search-duckdb: Executes a SQL query in-process using DuckDB. 

7""" 

8 

9from __future__ import annotations 

10 

11import json 

12from typing import Any 

13 

14import lumen.analytics.in_process as analytics 

15 

16from ...context import PluginContext 

17from ...operation import OperationInvocation 

18from ...plugin import Plugin, operation 

19 

20 

21class AnalyticsPlugin(Plugin): 

22 name = "analytics-plugin" 

23 version = "1.0.0" 

24 requires_cortex = "1.0" 

25 

26 @operation("search-pandas", on="Observation", scope="type", timeout_ms=30_000) 

27 async def search_pandas( 

28 self, ctx: PluginContext, op: OperationInvocation 

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

30 """FHIR Operation: Observation/$search-pandas 

31 

32 Fetches matching resources in-process and returns a flat JSON record list. 

33 """ 

34 df = await analytics.search_pandas("Observation", params=op.query) 

35 if df.empty: 

36 return [] 

37 return df.to_dict(orient="records") 

38 

39 @operation("search-duckdb", on="Observation", scope="type", timeout_ms=30_000) 

40 async def search_duckdb( 

41 self, ctx: PluginContext, op: OperationInvocation 

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

43 """FHIR Operation: Observation/$search-duckdb 

44 

45 Executes a DuckDB SQL query on the server-side DataFrame. 

46 """ 

47 import duckdb 

48 

49 body = op.body or {} 

50 if isinstance(body, bytes): 

51 try: 

52 body = json.loads(body.decode("utf-8")) 

53 except Exception: 

54 body = {} 

55 

56 sql_query = body.get("query") 

57 if not sql_query: 

58 return [{"error": "Missing 'query' parameter in request body"}] 

59 

60 fhir_params = body.get("params", {}) 

61 

62 # Load data in-process as a Pandas DataFrame 

63 df = await analytics.search_pandas("Observation", params=fhir_params) 

64 if df.empty: 

65 return [] 

66 

67 # Run DuckDB SQL query in-process on the DataFrame 

68 try: 

69 conn = duckdb.connect() 

70 conn.register("df", df) 

71 conn.register("Observation", df) 

72 result_df = conn.execute(sql_query).df() 

73 except Exception as e: 

74 return [{"error": f"DuckDB SQL query failed: {e!s}"}] 

75 

76 return result_df.to_dict(orient="records")