Coverage for src/ezcompiler/services/uploader_service.py: 52.08%
78 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-04 11:22 +0000
1# ///////////////////////////////////////////////////////////////
2# UPLOADER_SERVICE - Upload orchestration service
3# Project: ezcompiler
4# ///////////////////////////////////////////////////////////////
6"""
7Uploader service - Upload orchestration service for EzCompiler.
9This module provides the UploaderService class that orchestrates file
10and directory uploads using different upload backends (disk, server).
12Services layer can use WARNING and ERROR log levels.
13"""
15from __future__ import annotations
17# ///////////////////////////////////////////////////////////////
18# IMPORTS
19# ///////////////////////////////////////////////////////////////
20# Standard library imports
21from pathlib import Path
22from typing import TYPE_CHECKING, Any, Literal
24# Local imports
25from ..adapters import UploaderFactory
26from ..shared._constants import RELEASE_SUBDIR, UPDATE_SUBDIR
27from ..shared.exceptions import UploadError
28from ..utils.validators import validate_upload_structure
30if TYPE_CHECKING: 30 ↛ 31line 30 didn't jump to line 31 because the condition on line 30 was never true
31 from .._types import UploaderPort
32 from ..shared._compiler_config import CompilerConfig
34# ///////////////////////////////////////////////////////////////
35# TYPE ALIASES
36# ///////////////////////////////////////////////////////////////
38UploadType = Literal["disk", "server", "r2"]
40# ///////////////////////////////////////////////////////////////
41# CLASSES
42# ///////////////////////////////////////////////////////////////
45class UploaderService:
46 """
47 Upload orchestration service.
49 Orchestrates file and directory uploads using different upload backends.
50 Handles upload type selection, validation, and execution.
52 Example:
53 >>> service = UploaderService()
54 >>> service.upload(
55 ... source_path=Path("dist.zip"),
56 ... upload_type="disk",
57 ... destination="releases/",
58 ... upload_config={"overwrite": True}
59 ... )
60 """
62 # ////////////////////////////////////////////////
63 # UPLOAD METHODS
64 # ////////////////////////////////////////////////
66 @staticmethod
67 def upload(
68 source_path: Path,
69 upload_type: UploadType,
70 destination: str,
71 upload_config: dict[str, Any] | None = None,
72 ) -> None:
73 """
74 Upload a file or directory to the specified destination.
76 Args:
77 source_path: Path to the source file or directory
78 upload_type: Type of upload ("disk" or "server")
79 destination: Destination path or URL
80 upload_config: Additional uploader configuration options
82 Raises:
83 UploadError: If upload fails or upload type is invalid
85 Example:
86 >>> UploaderService.upload(
87 ... Path("dist.zip"),
88 ... "disk",
89 ... "releases/",
90 ... {"overwrite": True}
91 ... )
92 """
93 try:
94 # Validate upload type
95 if not validate_upload_structure(upload_type):
96 raise UploadError(f"Invalid upload type: {upload_type}")
98 # Prepare uploader configuration
99 config = upload_config or {}
100 if upload_type == "disk":
101 config["destination_path"] = destination
102 elif upload_type == "server":
103 config["server_url"] = destination
104 # r2: bucket vient de upload_config, destination = préfixe objet
106 # Create uploader and perform upload
107 uploader: UploaderPort = UploaderFactory.create_uploader(
108 upload_type, config
109 )
110 uploader.upload(source_path=source_path, destination=destination)
111 except UploadError:
112 raise
113 except Exception as e:
114 raise UploadError(f"Upload failed: {str(e)}") from e
116 @staticmethod
117 def download(
118 remote_source: str,
119 upload_type: UploadType,
120 destination_local: Path,
121 upload_config: dict[str, Any] | None = None,
122 ) -> None:
123 """Download a remote tree into ``destination_local`` via the backend.
125 Args:
126 remote_source: Remote source (path, URL or prefix) to fetch.
127 upload_type: Backend type ("disk", "server" or "r2").
128 destination_local: Local directory to populate.
129 upload_config: Additional uploader configuration options.
131 Raises:
132 UploadError: If the download fails or the type is invalid.
133 """
134 try:
135 uploader: UploaderPort = UploaderFactory.create_uploader(
136 upload_type, upload_config or {}
137 )
138 uploader.download(remote_source, destination_local)
139 except UploadError:
140 raise
141 except Exception as e:
142 raise UploadError(f"Download failed: {str(e)}") from e
144 @staticmethod
145 def upload_release(
146 config: CompilerConfig,
147 repo_dir: Path,
148 release_root: Path | None,
149 destination: str | None = None,
150 repo_destination: str | None = None,
151 release_destination: str | None = None,
152 upload_config: dict[str, Any] | None = None,
153 ) -> None:
154 """Double upload séquentiel : arbre TUF puis zip installeur.
156 Étape 1 — upload arbre TUF vers ``<dest>/update/`` (ou préfixe R2).
157 Étape 2 — upload zip installeur vers ``<dest>/release/`` (ignoré si R2).
159 Note: Le double-upload n'est pas atomique. Si l'upload du zip échoue,
160 le repo TUF est déjà en ligne. En cas d'échec, ré-exécuter upload()
161 pour reprendre.
163 Args:
164 config: CompilerConfig contenant les destinations et options R2.
165 repo_dir: Répertoire local du repo TUF.
166 release_root: Répertoire local du zip installeur (None si R2).
167 destination: Override commun pour les deux destinations.
168 repo_destination: Override de ``config.repo_destination``.
169 release_destination: Override de ``config.release_destination``.
170 upload_config: Options supplémentaires passées aux uploaders.
172 Raises:
173 UploadError: Si un upload échoue.
174 """
175 repo_dest = repo_destination or config.repo_destination
176 rel_dest = release_destination or config.release_destination
178 UploaderService._upload_tuf_repo(
179 config, repo_dir, repo_dest, destination, upload_config
180 )
182 if release_root is not None:
183 UploaderService._upload_release_zip(
184 config, release_root, rel_dest, destination, upload_config
185 )
187 @staticmethod
188 def _upload_tuf_repo(
189 config: CompilerConfig,
190 repo_dir: Path,
191 repo_dest: str,
192 destination: str | None,
193 upload_config: dict[str, Any] | None,
194 ) -> None:
195 """Upload l'arbre TUF vers la destination configurée."""
196 try:
197 if repo_dest == "r2":
198 endpoint = config.repo_endpoint
199 bucket, _, prefix = endpoint.partition("/")
200 UploaderService.upload(
201 source_path=repo_dir,
202 upload_type="r2",
203 destination=prefix,
204 upload_config={"bucket": bucket},
205 )
206 elif repo_dest == "server": 206 ↛ 207line 206 didn't jump to line 207 because the condition on line 206 was never true
207 base = destination or config.resolved_repo_destination or ""
208 UploaderService.upload(
209 source_path=repo_dir,
210 upload_type="server",
211 destination=base.rstrip("/") + f"/{UPDATE_SUBDIR}",
212 upload_config=upload_config,
213 )
214 else: # disk (default)
215 base = destination or config.resolved_repo_destination or ""
216 UploaderService.upload(
217 source_path=repo_dir,
218 upload_type="disk",
219 destination=str(Path(base) / UPDATE_SUBDIR),
220 upload_config=upload_config,
221 )
222 except UploadError as e:
223 raise UploadError(f"TUF repo upload failed: {e}") from e
225 @staticmethod
226 def _upload_release_zip(
227 config: CompilerConfig,
228 release_root: Path,
229 rel_dest: str,
230 destination: str | None,
231 upload_config: dict[str, Any] | None,
232 ) -> None:
233 """Upload le zip installeur vers la destination configurée."""
234 try:
235 if rel_dest == "r2": 235 ↛ 236line 235 didn't jump to line 236 because the condition on line 235 was never true
236 endpoint = config.release_endpoint
237 bucket, _, prefix = endpoint.partition("/")
238 UploaderService.upload(
239 source_path=release_root,
240 upload_type="r2",
241 destination=prefix,
242 upload_config={"bucket": bucket},
243 )
244 elif rel_dest == "server": 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true
245 base = destination or config.resolved_release_destination or ""
246 UploaderService.upload(
247 source_path=release_root,
248 upload_type="server",
249 destination=base.rstrip("/") + f"/{RELEASE_SUBDIR}",
250 upload_config=upload_config,
251 )
252 else: # disk (default)
253 base = destination or config.resolved_release_destination or ""
254 UploaderService.upload(
255 source_path=release_root,
256 upload_type="disk",
257 destination=str(Path(base) / RELEASE_SUBDIR),
258 upload_config=upload_config,
259 )
260 except UploadError as e:
261 raise UploadError(f"Release zip upload failed: {e}") from e
263 # ////////////////////////////////////////////////
264 # UTILITY METHODS
265 # ////////////////////////////////////////////////
267 @staticmethod
268 def get_supported_types() -> list[str]:
269 """
270 Get list of supported upload types.
272 Returns:
273 list[str]: List of supported upload type names
275 Example:
276 >>> types = UploaderService.get_supported_types()
277 >>> print(types)
278 ['disk', 'server']
279 """
280 return UploaderFactory.get_supported_types()
282 @staticmethod
283 def validate_upload_config(
284 upload_type: UploadType, config: dict[str, Any] | None = None
285 ) -> bool:
286 """
287 Validate configuration for a specific upload type.
289 Args:
290 upload_type: Type of uploader to validate
291 config: Configuration to validate (default: None)
293 Returns:
294 bool: True if configuration is valid, False otherwise
296 Example:
297 >>> is_valid = UploaderService.validate_upload_config(
298 ... "disk", {"overwrite": True}
299 ... )
300 """
301 return UploaderFactory.validate_config(upload_type, config)