"""Provider-neutral contracts for private ARR object storage.""" from __future__ import annotations import re from dataclasses import dataclass from pathlib import PurePosixPath from typing import BinaryIO, Mapping, Optional, Protocol from arr_ingestion.contracts import ( ARTIFACT_SIZE_LIMITS, OPAQUE_ID_RE, ROLE_CONTRACTS, ArtifactRef, IngestionError, ) OBJECT_METADATA_SCHEMA = "1.0" OBJECT_STATES = frozenset({"staged", "committed"}) CANONICAL_OBJECT_FILENAMES = { "source_xml": "source.xml", "booking_source": "booking-source.xlsx", "daily_report": "daily-report.xlsx", "monthly_report": "monthly-report.xlsx", "company_report": "company-report.xlsx", "result_json": "result.json", "structured_result_json": "structured-result.json", "exception_report": "exception-report.xlsx", } PREFIX_SEGMENT_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,62}$") def valid_object_key(value: object) -> bool: if not isinstance(value, str) or not value or len(value) > 1024: return False if ( value.startswith("/") or value.endswith("/") or "\\" in value or "//" in value or any(ord(char) < 32 or ord(char) == 127 for char in value) ): return False path = PurePosixPath(value) return not path.is_absolute() and "." not in path.parts and ".." not in path.parts def valid_delivery_filename(role: str, value: object) -> bool: if role not in ROLE_CONTRACTS or not isinstance(value, str): return False _file_kind, extension, _mime_type = ROLE_CONTRACTS[role] return ( 0 < len(value) <= 255 and PurePosixPath(value).name == value and value not in {".", ".."} and "/" not in value and "\\" not in value and not any(ord(char) < 32 or ord(char) == 127 for char in value) and value.lower().endswith(extension) ) @dataclass(frozen=True) class ObjectAddress: job_id: str attempt_no: int state: str role: str class ObjectKeyPolicy: """Builds and parses the only accepted ARR processing object-key shape.""" def __init__(self, prefix: str = "arr") -> None: parts = tuple(prefix.split("/")) if not parts or any(not PREFIX_SEGMENT_RE.fullmatch(part) for part in parts): raise ValueError("object prefix is invalid") self._parts = parts self.prefix = "/".join(parts) def build(self, address: ObjectAddress) -> str: self._validate_address(address) filename = CANONICAL_OBJECT_FILENAMES[address.role] return "/".join( self._parts + ( "jobs", address.job_id, "attempts", f"{address.attempt_no:04d}", address.state, address.role, filename, ) ) def parse(self, object_key: str) -> ObjectAddress: if not valid_object_key(object_key): raise IngestionError("ARTIFACT_REFERENCE_INVALID", "object key is invalid") parts = tuple(object_key.split("/")) tail = parts[len(self._parts) :] if parts[: len(self._parts)] != self._parts or len(tail) != 7: raise IngestionError("ARTIFACT_REFERENCE_INVALID", "object key is invalid") jobs, job_id, attempts, attempt_text, state, role, filename = tail if jobs != "jobs" or attempts != "attempts" or not attempt_text.isdigit(): raise IngestionError("ARTIFACT_REFERENCE_INVALID", "object key is invalid") attempt_no = int(attempt_text) address = ObjectAddress(job_id, attempt_no, state, role) self._validate_address(address) if attempt_text != f"{attempt_no:04d}" or filename != CANONICAL_OBJECT_FILENAMES[role]: raise IngestionError("ARTIFACT_REFERENCE_INVALID", "object key is invalid") return address @staticmethod def _validate_address(address: ObjectAddress) -> None: if ( not isinstance(address.job_id, str) or not OPAQUE_ID_RE.fullmatch(address.job_id) or not isinstance(address.attempt_no, int) or isinstance(address.attempt_no, bool) or not 1 <= address.attempt_no <= 9999 or address.state not in OBJECT_STATES or address.role not in CANONICAL_OBJECT_FILENAMES ): raise IngestionError("ARTIFACT_REFERENCE_INVALID", "object address is invalid") @dataclass(frozen=True) class BackendObject: object_key: str byte_size: int metadata: Mapping[str, str] etag: Optional[str] = None version_id: Optional[str] = None class BackendError(RuntimeError): """Normalized backend failure that never includes provider response text.""" def __init__(self, kind: str): super().__init__(kind) self.kind = kind class ObjectBackend(Protocol): """Adapter seam implemented by a filesystem, OSS, or S3-compatible client.""" def put_file( self, object_key: str, source: str, mime_type: str, metadata: Mapping[str, str], *, if_absent: bool, ) -> BackendObject: ... def head(self, object_key: str) -> BackendObject: ... def open_reader(self, object_key: str) -> BinaryIO: ... def copy_object( self, source_key: str, destination_key: str, metadata: Mapping[str, str], *, if_absent: bool, ) -> BackendObject: ... def delete_object(self, object_key: str) -> None: ... @dataclass(frozen=True) class StoredObject: object_key: str job_id: str attempt_no: int state: str role: str original_filename: str sha256: str byte_size: int mime_type: str etag: Optional[str] = None version_id: Optional[str] = None def to_artifact_ref(self) -> ArtifactRef: if self.state != "committed": raise IngestionError( "ARTIFACT_NOT_COMMITTED", "only committed objects may enter a delivery" ) file_kind, _extension, _mime_type = ROLE_CONTRACTS[self.role] return ArtifactRef( role=self.role, file_kind=file_kind, object_key=self.object_key, original_filename=self.original_filename, sha256=self.sha256, byte_size=self.byte_size, mime_type=self.mime_type, ) def role_limit(role: str) -> int: try: return ARTIFACT_SIZE_LIMITS[role] except KeyError: raise IngestionError("ARTIFACT_REFERENCE_INVALID", "artifact role is invalid") from None