Coverage for src/lexigram/admin/multitenancy/data_source.py: 41%

1"""TenantScopedDataSource — data source wrapper that injects tenant_id into queries.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.logging import get_logger 

8 

9logger = get_logger(__name__) 

10 

11 

12class TenantScopedDataSource: 

13 """Wraps an :class:`~lexigram.admin.data.data_source.IDataSource` and 

14 injects ``tenant_id`` into every query. 

15 

16 This ensures resources are automatically isolated per tenant without 

17 requiring every query to manually add the tenant filter. 

18 

19 Args: 

20 data_source: The underlying data source to delegate to. 

21 tenant_id: Active tenant identifier. 

22 tenant_field: Column/field name to filter on (default ``"tenant_id"``). 

23 

24 Example:: 

25 

26 scoped = TenantScopedDataSource(data_source, tenant_id="acme") 

27 result = await scoped.list(query) # → automatically adds tenant_id="acme" filter 

28 """ 

29 

30 def __init__( 

31 self, 

32 data_source: Any, 

33 tenant_id: str, 

34 tenant_field: str = "tenant_id", 

35 ) -> None: 

36 self._ds = data_source 

37 self._tenant_id = tenant_id 

38 self._tenant_field = tenant_field 

39 

40 def _inject_tenant(self, query: Any) -> Any: 

41 """Add the tenant filter to a query spec. 

42 

43 Works with :class:`~lexigram.admin.data.query.QuerySpec` objects that 

44 accept ``add_filter(field, op, value)``. If the query object doesn't 

45 support this, it is returned unchanged and a warning is logged. 

46 """ 

47 try: 

48 query.add_filter(self._tenant_field, "eq", self._tenant_id) 

49 except Exception: # noqa: BLE001 

50 logger.warning( 

51 "TenantScopedDataSource could not inject tenant filter into query %r", 

52 type(query).__name__, 

53 ) 

54 return query 

55 

56 async def list(self, query: Any) -> Any: 

57 """List records filtered to the active tenant.""" 

58 return await self._ds.list(self._inject_tenant(query)) 

59 

60 async def find_one(self, id: Any) -> Any: 

61 """Find a single record, ensuring it belongs to the active tenant.""" 

62 record = await self._ds.find_one(id) 

63 if record is None: 

64 return None 

65 record_tenant = getattr(record, self._tenant_field, None) or ( 

66 record.get(self._tenant_field) if isinstance(record, dict) else None 

67 ) 

68 if record_tenant and record_tenant != self._tenant_id: 

69 logger.warning( 

70 "TenantScopedDataSource: tenant mismatch for id=%s (expected %s, got %s)", 

71 id, 

72 self._tenant_id, 

73 record_tenant, 

74 ) 

75 return None 

76 return record 

77 

78 async def create(self, data: dict[str, Any]) -> Any: 

79 """Create a record with the tenant field pre-populated.""" 

80 data = {**data, self._tenant_field: self._tenant_id} 

81 return await self._ds.create(data) 

82 

83 async def update(self, id: Any, data: dict[str, Any]) -> Any: 

84 """Update a record that belongs to the active tenant.""" 

85 return await self._ds.update(id, data) 

86 

87 async def delete(self, id: Any) -> bool: 

88 """Delete a record that belongs to the active tenant.""" 

89 return await self._ds.delete(id) 

90 

91 @property 

92 def tenant_id(self) -> str: 

93 """The active tenant identifier.""" 

94 return self._tenant_id 

95 

96 

97__all__ = [ 

98 "TenantScopedDataSource", 

99]