Coverage for src/lexigram/admin/integrations/storage.py: 0%

1"""Storage integration — delegates file field storage to a blob store.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7if TYPE_CHECKING: 

8 from lexigram.contracts.core.di import ( 

9 ContainerRegistrarProtocol, 

10 ContainerResolverProtocol, 

11 ) 

12 

13 

14class _NoOpStorage: 

15 async def put( 

16 self, path: str, data: bytes, content_type: str = "application/octet-stream" 

17 ) -> dict[str, Any]: 

18 return {"path": path, "size": len(data)} 

19 

20 async def get(self, path: str) -> bytes: 

21 return b"" 

22 

23 async def delete(self, path: str) -> bool: 

24 return True 

25 

26 async def presigned_url(self, path: str, expires_in: int = 3600) -> str: 

27 return "" 

28 

29 

30class StorageIntegration: 

31 """Adapter that delegates file storage to lexigram-storage. 

32 

33 Gracefully no-ops when ``lexigram-storage`` is not installed or the 

34 integration is disabled. 

35 """ 

36 

37 def __init__(self, config: Any) -> None: 

38 self._config = config 

39 self._store: Any = None 

40 self._enabled = False 

41 

42 def register(self, container: ContainerRegistrarProtocol) -> None: 

43 from lexigram.admin.config import StorageIntegrationConfig 

44 from lexigram.admin.integrations._optional import is_installed 

45 

46 cfg = self._config 

47 if not isinstance(cfg, StorageIntegrationConfig): 

48 cfg = StorageIntegrationConfig() 

49 if not cfg.enabled: 

50 self._store = _NoOpStorage() 

51 return 

52 if not is_installed("lexigram.storage"): 

53 self._store = _NoOpStorage() 

54 return 

55 self._enabled = True 

56 

57 async def boot(self, container: ContainerResolverProtocol) -> None: 

58 if not self._enabled: 

59 return 

60 try: 

61 from lexigram.contracts.infra.storage import BlobStoreProtocol 

62 

63 self._store = await container.resolve(BlobStoreProtocol) 

64 except Exception: # noqa: BLE001 

65 self._store = _NoOpStorage() 

66 

67 async def shutdown(self) -> None: 

68 pass 

69 

70 async def health_check(self) -> dict[str, Any]: 

71 return { 

72 "status": "healthy" if not isinstance(self._store, _NoOpStorage) else "noop" 

73 } 

74 

75 async def put( 

76 self, path: str, data: bytes, content_type: str = "application/octet-stream" 

77 ) -> dict[str, Any]: 

78 return await self._store.upload(path, data, content_type=content_type) 

79 

80 async def get(self, path: str) -> bytes: 

81 return await self._store.download(path) 

82 

83 async def delete(self, path: str) -> bool: 

84 return await self._store.delete(path) 

85 

86 async def presigned_url(self, path: str, expires_in: int | None = None) -> str: 

87 ttl = expires_in or getattr(self._config, "presigned_url_expiry", 3600) 

88 return await self._store.get_presigned_url(path, expires_in=ttl, method="get") 

89 

90 

91__all__ = ["StorageIntegration"]