Coverage for src/lumen/analytics/in_process.py: 71%

1""" 

2In-process analytics engine for Lumen. 

3 

4Executes queries directly against the database pool and ResourceService 

5in-process, returning zero-copy Pandas DataFrames without HTTP overhead. 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import Any 

11from uuid import UUID 

12 

13from ..api.deps import resolve_tenant_settings 

14from ..config import get_settings 

15from ..db.pool import get_pool 

16from ..service.resources import ResourceService 

17 

18 

19async def search_pandas( 

20 resource_type: str, 

21 params: dict[str, list[str]] | None = None, 

22 tenant_id: UUID | None = None, 

23 **kwargs: Any, 

24) -> Any: 

25 """Execute an in-process search and return a flat Pandas DataFrame. 

26 

27 If ``tenant_id`` is omitted, the default tenant configured in ``Settings`` 

28 is used. 

29 """ 

30 settings = get_settings() 

31 tenant = tenant_id or settings.default_tenant 

32 tenant_settings = await resolve_tenant_settings(tenant, settings) 

33 pool = get_pool() 

34 

35 service = ResourceService(tenant, tenant_settings=tenant_settings, pool=pool) 

36 return await service.search_pandas(resource_type, params=params, **kwargs) 

37 

38 

39async def search_arrow( 

40 resource_type: str, 

41 params: dict[str, list[str]] | None = None, 

42 tenant_id: UUID | None = None, 

43 **kwargs: Any, 

44) -> Any: 

45 """Execute an in-process search and return an Apache Arrow Table. 

46 

47 If ``tenant_id`` is omitted, the default tenant configured in ``Settings`` 

48 is used. 

49 """ 

50 settings = get_settings() 

51 tenant = tenant_id or settings.default_tenant 

52 tenant_settings = await resolve_tenant_settings(tenant, settings) 

53 pool = get_pool() 

54 

55 service = ResourceService(tenant, tenant_settings=tenant_settings, pool=pool) 

56 return await service.search_arrow(resource_type, params=params, **kwargs)