Coverage for src / lexigram / contracts / infra / storage / protocols.py: 0%

1"""Storage blob protocol class definitions.""" 

2 

3from __future__ import annotations 

4 

5from datetime import timedelta 

6from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

7 

8from lexigram.contracts.core.provider import ProviderProtocol 

9from lexigram.contracts.infra.storage.models import FileInfo 

10 

11if TYPE_CHECKING: 

12 from collections.abc import AsyncIterator 

13 

14 from lexigram.contracts.core import HealthCheckResult 

15 

16 

17@runtime_checkable 

18class BlobStoreProtocol(Protocol): 

19 """Protocol for blob storage operations. 

20 

21 This interface defines the contract for file storage backends 

22 (S3, GCS, Azure Blob, local filesystem, etc.). 

23 

24 Example: 

25 ```python 

26 class S3BlobStore: 

27 async def upload(self, path: str, data: bytes, **options) -> FileInfo: 

28 await self._client.put_object(Bucket=self._bucket, Key=path, Body=data) 

29 return FileInfo(path=path, size=len(data), ...) 

30 ``` 

31 """ 

32 

33 async def upload( 

34 self, 

35 path: str, 

36 data: bytes | AsyncIterator[bytes], 

37 content_type: str | None = None, 

38 **options: Any, 

39 ) -> FileInfo: 

40 """Upload data to the storage backend. 

41 

42 Args: 

43 path: Storage path/key. 

44 data: File content as bytes or async iterator. 

45 content_type: MIME type of the content. 

46 **options: Additional upload options. 

47 

48 Returns: 

49 FileInfo with path, size, and metadata. 

50 """ 

51 ... 

52 

53 async def download(self, path: str) -> bytes: 

54 """Download file content into memory. 

55 

56 Args: 

57 path: Storage path/key. 

58 

59 Returns: 

60 File content as bytes. 

61 """ 

62 ... 

63 

64 def stream(self, path: str, chunk_size: int = 8192) -> AsyncIterator[bytes]: 

65 """Stream file content (memory efficient). 

66 

67 Args: 

68 path: Storage path/key. 

69 chunk_size: Size of each chunk in bytes. 

70 

71 Yields: 

72 File content in chunks. 

73 """ 

74 ... 

75 

76 async def delete(self, path: str) -> None: 

77 """Delete a file. 

78 

79 Args: 

80 path: Storage path/key. 

81 """ 

82 ... 

83 

84 async def exists(self, path: str) -> bool: 

85 """Check if file exists. 

86 

87 Args: 

88 path: Storage path/key. 

89 

90 Returns: 

91 True if file exists. 

92 """ 

93 ... 

94 

95 async def info(self, path: str) -> FileInfo: 

96 """Get file metadata. 

97 

98 Args: 

99 path: Storage path/key. 

100 

101 Returns: 

102 FileInfo with size, content_type, etc. 

103 """ 

104 ... 

105 

106 def list(self, prefix: str = "") -> AsyncIterator[FileInfo]: 

107 """List files with a given prefix. 

108 

109 Args: 

110 prefix: Path prefix to filter by. 

111 

112 Yields: 

113 FileInfo for each matching file. 

114 """ 

115 ... 

116 

117 async def get_url(self, path: str) -> str: 

118 """Get public URL (if applicable). 

119 

120 Args: 

121 path: Storage path/key. 

122 

123 Returns: 

124 Public URL string. 

125 """ 

126 ... 

127 

128 async def get_presigned_url( 

129 self, 

130 path: str, 

131 expires_in: timedelta = timedelta(hours=1), 

132 method: str = "GET", 

133 ) -> str: 

134 """Get a temporary secure URL. 

135 

136 Args: 

137 path: Storage path/key. 

138 expires_in: URL validity window as a :class:`~datetime.timedelta` 

139 (default one hour). Pass ``timedelta(minutes=5)`` for secure 

140 short-lived downloads or ``timedelta(hours=24)`` for bulk exports. 

141 method: HTTP method (GET or PUT). 

142 

143 Returns: 

144 Presigned URL string. 

145 """ 

146 ... 

147 

148 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

149 """Perform health check. 

150 

151 Returns: 

152 Structured health check result. 

153 """ 

154 ... 

155 

156 

157@runtime_checkable 

158class StorageDriverProtocol(Protocol): 

159 """Protocol for storage driver implementations. 

160 

161 This interface extends BlobStoreProtocol with driver-specific operations 

162 such as copy, move, and batch operations. Storage drivers 

163 (S3, GCS, Azure, local filesystem, memory) should implement this 

164 protocol or extend BaseDriver. 

165 

166 Example: 

167 ```python 

168 class S3Driver: 

169 async def copy(self, source: str, destination: str) -> None: 

170 await self._client.copy_object(Bucket=self._bucket, Key=destination, CopySource=source) 

171 

172 async def move(self, source: str, destination: str) -> None: 

173 await self.copy(source, destination) 

174 await self.delete(source) 

175 ``` 

176 """ 

177 

178 async def copy(self, source: str, destination: str) -> None: 

179 """Copy a file from source to destination. 

180 

181 Args: 

182 source: Source storage path/key. 

183 destination: Destination storage path/key. 

184 """ 

185 ... 

186 

187 async def move(self, source: str, destination: str) -> None: 

188 """Move a file from source to destination. 

189 

190 Args: 

191 source: Source storage path/key. 

192 destination: Destination storage path/key. 

193 """ 

194 ... 

195 

196 async def copy_batch(self, operations: list[tuple[str, str]]) -> list[str]: 

197 """Execute batch copy operations. 

198 

199 Args: 

200 operations: List of (source, destination) tuples. 

201 

202 Returns: 

203 List of destination paths that were copied. 

204 """ 

205 ... 

206 

207 

208@runtime_checkable 

209class StorageProviderProtocol(ProviderProtocol, Protocol): 

210 """Protocol for storage providers. 

211 

212 Storage providers are responsible for setting up blob stores, 

213 file systems, and CDN integration. 

214 """ 

215 

216 

217__all__ = [ 

218 "BlobStoreProtocol", 

219 "FileInfo", 

220 "StorageDriverProtocol", 

221 "StorageProviderProtocol", 

222]