Coverage for fastblocks / adapters / images / cloudflare.py: 29%

1"""Cloudflare Images adapter for FastBlocks.""" 

2 

3import asyncio 

4from contextlib import suppress 

5from typing import Any 

6from uuid import UUID, uuid4 

7 

8import httpx 

9from acb.depends import depends 

10from pydantic import SecretStr 

11 

12from ._base import ImagesBase, ImagesBaseSettings 

13 

14 

15class CloudflareImagesSettings(ImagesBaseSettings): 

16 """Settings for Cloudflare Images adapter.""" 

17 

18 # Required ACB 0.19.0+ metadata 

19 MODULE_ID: UUID = UUID("01937d86-5e3b-7f4c-9e8e-a5b6c7d8e9f0") # Static UUID7 

20 MODULE_STATUS: str = "stable" 

21 

22 # Cloudflare API configuration 

23 account_id: str = "" 

24 api_token: SecretStr = SecretStr("") 

25 delivery_url: str = "" # https://imagedelivery.net/{account_hash} 

26 

27 # Image configuration 

28 default_variant: str = "public" 

29 require_signed_urls: bool = False 

30 timeout: int = 30 

31 

32 # R2 storage configuration (optional) 

33 r2_bucket: str | None = None 

34 r2_public_url: str | None = None 

35 

36 

37class CloudflareImages(ImagesBase): 

38 """Cloudflare Images adapter with R2 storage integration.""" 

39 

40 # Required ACB 0.19.0+ metadata 

41 MODULE_ID: UUID = UUID("01937d86-5e3b-7f4c-9e8e-a5b6c7d8e9f0") # Static UUID7 

42 MODULE_STATUS: str = "stable" 

43 

44 def __init__(self) -> None: 

45 """Initialize Cloudflare Images adapter.""" 

46 super().__init__() 

47 self.settings: CloudflareImagesSettings | None = None 

48 self._client: httpx.AsyncClient | None = None 

49 

50 # Register with ACB dependency system 

51 with suppress(Exception): 

52 depends.set(self) 

53 

54 async def _get_client(self) -> httpx.AsyncClient: 

55 """Get or create HTTP client.""" 

56 if not self._client: 

57 if not self.settings: 

58 self.settings = CloudflareImagesSettings() 

59 

60 self._client = httpx.AsyncClient( 

61 timeout=self.settings.timeout, 

62 headers={ 

63 "Authorization": f"Bearer {self.settings.api_token.get_secret_value()}", 

64 "Content-Type": "application/json", 

65 }, 

66 ) 

67 return self._client 

68 

69 async def upload_image(self, file_data: bytes, filename: str) -> str: 

70 """Upload image to Cloudflare Images.""" 

71 if not self.settings: 

72 self.settings = CloudflareImagesSettings() 

73 

74 client = await self._get_client() 

75 

76 # Generate unique ID for the image 

77 str(uuid4()) 

78 

79 # Prepare upload data 

80 

81 # Upload to Cloudflare Images 

82 url = f"https://api.cloudflare.com/client/v4/accounts/{self.settings.account_id}/images/v1" 

83 

84 # Format files for httpx - each value is a tuple of (filename, file_content, content_type) 

85 files_data: dict[str, tuple[str, bytes, str]] = { 

86 "file": (filename, file_data, "image/*"), 

87 } 

88 

89 response = await client.post(url, files=files_data) 

90 response.raise_for_status() 

91 

92 result = response.json() 

93 if not result.get("success"): 

94 raise RuntimeError( 

95 f"Upload failed: {result.get('errors', 'Unknown error')}" 

96 ) 

97 

98 # Return the image ID for future reference 

99 uploaded_image_id: str = result["result"]["id"] 

100 return uploaded_image_id 

101 

102 async def get_image_url( 

103 self, image_id: str, transformations: dict[str, Any] | None = None 

104 ) -> str: 

105 """Generate Cloudflare Images URL with transformations.""" 

106 if not self.settings: 

107 self.settings = CloudflareImagesSettings() 

108 

109 base_url = self._build_base_url(image_id) 

110 

111 if transformations: 

112 transform_parts = self._build_transformation_parts(transformations) 

113 if transform_parts: 

114 return self._build_transformed_url( 

115 base_url, transformations, transform_parts 

116 ) 

117 

118 return f"{base_url}/{self.settings.default_variant}" 

119 

120 def _build_base_url(self, image_id: str) -> str: 

121 """Build base URL for Cloudflare image.""" 

122 if not self.settings: 

123 raise RuntimeError("Cloudflare Images settings not configured") 

124 

125 if self.settings.delivery_url: 

126 return f"{self.settings.delivery_url}/{image_id}" 

127 return f"https://api.cloudflare.com/client/v4/accounts/{self.settings.account_id}/images/v1/{image_id}" 

128 

129 @staticmethod 

130 def _build_transformation_parts(transformations: dict[str, Any]) -> list[str]: 

131 """Build transformation query parameters.""" 

132 # Common transformations 

133 common_parts = [ 

134 f"{key}={transformations[key]}" 

135 for key in ("width", "height", "quality", "format", "fit") 

136 if key in transformations 

137 ] 

138 

139 # Advanced transformations 

140 advanced_parts = [ 

141 f"{key}={transformations[key]}" 

142 for key in ("blur", "brightness", "contrast", "gamma", "sharpen") 

143 if key in transformations 

144 ] 

145 

146 return common_parts + advanced_parts 

147 

148 def _build_transformed_url( 

149 self, base_url: str, transformations: dict[str, Any], transform_parts: list[str] 

150 ) -> str: 

151 """Build final URL with transformations.""" 

152 if not self.settings: 

153 raise RuntimeError("Cloudflare Images settings not configured") 

154 

155 variant = transformations.get("variant", self.settings.default_variant) 

156 transform_string = ",".join(transform_parts) 

157 return f"{base_url}/{variant}?{transform_string}" 

158 

159 def get_img_tag(self, image_id: str, alt: str, **attributes: Any) -> str: 

160 """Generate img tag with Cloudflare Images patterns.""" 

161 # Generate base URL 

162 transformations = attributes.pop("transformations", {}) 

163 

164 # Use async context for URL generation (in real usage) 

165 try: 

166 loop = asyncio.get_event_loop() 

167 url = loop.run_until_complete(self.get_image_url(image_id, transformations)) 

168 except RuntimeError: 

169 # Create new event loop if none exists 

170 loop = asyncio.new_event_loop() 

171 asyncio.set_event_loop(loop) 

172 url = loop.run_until_complete(self.get_image_url(image_id, transformations)) 

173 

174 # Build attributes 

175 img_attrs = { 

176 "src": url, 

177 "alt": alt, 

178 "loading": "lazy" 

179 if self.settings and self.settings.lazy_loading 

180 else "eager", 

181 } | attributes 

182 

183 # Generate tag 

184 attr_string = " ".join( 

185 f'{k}="{v}"' for k, v in img_attrs.items() if v is not None 

186 ) 

187 return f"<img {attr_string}>" 

188 

189 async def delete_image(self, image_id: str) -> bool: 

190 """Delete image from Cloudflare Images.""" 

191 if not self.settings: 

192 self.settings = CloudflareImagesSettings() 

193 

194 client = await self._get_client() 

195 

196 url = f"https://api.cloudflare.com/client/v4/accounts/{self.settings.account_id}/images/v1/{image_id}" 

197 

198 response = await client.delete(url) 

199 response.raise_for_status() 

200 

201 result = response.json() 

202 success: bool = result.get("success", False) 

203 return success 

204 

205 async def list_images(self, page: int = 1, per_page: int = 50) -> dict[str, Any]: 

206 """List images in Cloudflare Images.""" 

207 if not self.settings: 

208 self.settings = CloudflareImagesSettings() 

209 

210 client = await self._get_client() 

211 

212 url = f"https://api.cloudflare.com/client/v4/accounts/{self.settings.account_id}/images/v1" 

213 params = {"page": page, "per_page": per_page} 

214 

215 response = await client.get(url, params=params) 

216 response.raise_for_status() 

217 

218 result: dict[str, Any] = response.json() 

219 return result 

220 

221 async def get_usage_stats(self) -> dict[str, Any]: 

222 """Get Cloudflare Images usage statistics.""" 

223 if not self.settings: 

224 self.settings = CloudflareImagesSettings() 

225 

226 client = await self._get_client() 

227 

228 url = f"https://api.cloudflare.com/client/v4/accounts/{self.settings.account_id}/images/v1/stats" 

229 

230 response = await client.get(url) 

231 response.raise_for_status() 

232 

233 stats: dict[str, Any] = response.json() 

234 return stats 

235 

236 async def close(self) -> None: 

237 """Close HTTP client.""" 

238 if self._client: 

239 await self._client.aclose() 

240 self._client = None 

241 

242 

243# Template filter registration for FastBlocks 

244def register_cloudflare_filters(env: Any) -> None: 

245 """Register Cloudflare Images filters for Jinja2 templates.""" 

246 

247 @env.filter("cf_image_url") # type: ignore[misc] 

248 async def cf_image_url_filter(image_id: str, **transformations: Any) -> str: 

249 """Template filter for Cloudflare Images URLs.""" 

250 images = await depends.get("images") 

251 if isinstance(images, CloudflareImages): 

252 return await images.get_image_url(image_id, transformations) 

253 return f"#{image_id}" # Fallback 

254 

255 @env.filter("cf_img_tag") # type: ignore[misc] 

256 def cf_img_tag_filter(image_id: str, alt: str = "", **attributes: Any) -> str: 

257 """Template filter for complete Cloudflare img tags.""" 

258 images = depends.get_sync("images") 

259 if isinstance(images, CloudflareImages): 

260 return images.get_img_tag(image_id, alt, **attributes) 

261 return f'<img src="#{image_id}" alt="{alt}">' # Fallback 

262 

263 @env.global_("cloudflare_responsive_img") # type: ignore[misc] 

264 def cloudflare_responsive_img( 

265 image_id: str, 

266 alt: str, 

267 sizes: str = "(max-width: 768px) 100vw, 50vw", 

268 **attributes: Any, 

269 ) -> str: 

270 """Generate responsive image with multiple sizes.""" 

271 images = depends.get_sync("images") 

272 if not isinstance(images, CloudflareImages): 

273 return f'<img src="#{image_id}" alt="{alt}">' 

274 

275 # Generate srcset for different screen sizes 

276 srcset_parts = [] 

277 widths = [320, 640, 768, 1024, 1280, 1536] 

278 

279 for width in widths: 

280 try: 

281 loop = asyncio.get_event_loop() 

282 url = loop.run_until_complete( 

283 images.get_image_url( 

284 image_id, {"width": width, "fit": "scale-down"} 

285 ) 

286 ) 

287 srcset_parts.append(f"{url} {width}w") 

288 except Exception: 

289 continue 

290 

291 # Build img tag with srcset 

292 base_url = asyncio.get_event_loop().run_until_complete( 

293 images.get_image_url(image_id, {"width": 1024, "fit": "scale-down"}) 

294 ) 

295 

296 img_attrs = { 

297 "src": base_url, 

298 "alt": alt, 

299 "sizes": sizes, 

300 "srcset": ", ".join(srcset_parts), 

301 "loading": "lazy", 

302 } | attributes 

303 

304 attr_string = " ".join( 

305 f'{k}="{v}"' for k, v in img_attrs.items() if v is not None 

306 ) 

307 return f"<img {attr_string}>" 

308 

309 

310ImagesSettings = CloudflareImagesSettings 

311Images = CloudflareImages 

312 

313 

314# ACB 0.19.0+ compatibility 

315__all__ = [ 

316 "CloudflareImages", 

317 "CloudflareImagesSettings", 

318 "register_cloudflare_filters", 

319 "Images", 

320 "ImagesSettings", 

321]