"""Services for uploading rasters using STAC naming conventions."""
from __future__ import annotations
from pathlib import Path
from typing import Optional, Union
from uuid import uuid4
from ...storage import CloudObjectStore, StorageError
from .base import AbstractRasterUploadHandler
from .default import DefaultRasterUploadHandler
from .filename import DateInput, build_stac_filename
LocalPath = Union[str, Path]
[docs]
class RasterUploadService:
"""Coordinates filename generation, path building, and uploads."""
def __init__(self, handler: Optional[AbstractRasterUploadHandler] = None) -> None:
self._handler: AbstractRasterUploadHandler = handler or DefaultRasterUploadHandler()
[docs]
def set_handler(self, handler: AbstractRasterUploadHandler) -> None:
if handler is None:
raise StorageError("upload handler cannot be None")
self._handler = handler
[docs]
def build_remote_path(
self,
collection: str,
filename: str,
subfolder: Optional[str] = None,
) -> str:
return self._handler.build_remote_path(collection, filename, subfolder)
[docs]
def upload_file(
self,
storage: CloudObjectStore,
local_path: LocalPath,
*,
collection: str,
filename: str,
subfolder: Optional[str] = None,
overwrite: bool = False,
ensure_unique: bool = True,
) -> str:
remote_path = self.build_remote_path(collection, filename, subfolder)
remote_path = self._append_uuid_suffix(remote_path, ensure_unique)
return storage.upload_file(
local_path,
remote_path=remote_path,
overwrite=overwrite,
)
[docs]
def upload_raster_asset(
self,
storage: CloudObjectStore,
local_path: LocalPath,
*,
collection: str,
acquisition_date: DateInput,
tile: Optional[str] = None,
subfolder: Optional[str] = None,
overwrite: bool = False,
ensure_unique: bool = True,
) -> str:
filename = build_stac_filename(acquisition_date, tile)
return self.upload_file(
storage,
local_path,
collection=collection,
filename=filename,
subfolder=subfolder,
overwrite=overwrite,
ensure_unique=ensure_unique,
)
def _append_uuid_suffix(self, remote_path: str, ensure_unique: bool) -> str:
"""Append a UUID suffix to the remote path if ensure_unique is True."""
if not ensure_unique or not remote_path:
return remote_path
parts = remote_path.rsplit("/", 1)
if len(parts) == 2:
prefix, name = parts
else:
prefix, name = "", parts[0]
if "." in name:
stem, ext = name.rsplit(".", 1)
ext = f".{ext}"
else:
stem, ext = name, ""
new_name = f"{stem}_{uuid4().hex}{ext}"
return f"{prefix}/{new_name}" if prefix else new_name