API Reference — Aquilia Documentation
Comprehensive guide and documentation for API Reference in the Aquilia framework. View API reference, examples, and implementation patterns.
Unified Storage / API Reference Storage API Reference Complete interface contract specifications for the unified storage registry, backend drivers, metadata structures, and async file handles. Table of Contents , , , , , ].map((item, i) => ( • ))} StorageRegistry The central coordinator for registering, retrieving, and checking the health of multiple storage backends. class StorageRegistry: def __init__(self) -> None: """Create an empty storage registry.""" @property def default(self) -> StorageBackend: """Get the default registered StorageBackend instance. Raises StorageConfigFault if no default backend is set. """ def register(self, alias: str, backend: StorageBackend) -> None: """Register a backend instance with the given alias name.""" def unregister(self, alias: str) -> None: """Unregister a backend by its alias name.""" def set_default(self, alias: str) -> None: """Define which registered alias acts as the default backend.""" def get(self, alias: str) -> StorageBackend | None: """Look up a backend instance by alias. Returns None if missing.""" def __getitem__(self, alias: str) -> StorageBackend: """Access a backend via bracket notation registry[alias]. Raises KeyError if alias is not found. """ async def initialize_all(self) -> None: """Run the async initialize() lifecycle hook on all registered backends.""" async def shutdown_all(self) -> None: """Close/release connections on all registered backends.""" async def health_check(self) -> dict[str, bool]: """Runs a ping() check on each registered backend. Returns a dictionary of . """ StorageBackend Abstract Base Class establishing the contract all storage drivers (Local, S3, Memory, GCS, SFTP) must implement. from abc import ABC, abstractmethod from collections.abc import AsyncIterator from typing import BinaryIO class StorageBackend(ABC): @property @abstractmethod def backend_name(self) -> str: """Returns the driver name identifier (e.g. 'local', 's3').""" async def initialize(self) -> None: """Bootstrap directory structures, connections, or credential handshakes.""" async def ping(self) -> bool: """Verifies driver accessibility. Returns True if healthy, False otherwise.""" @abstractmethod async def save( self, name: str, content: bytes | BinaryIO | AsyncIterator[bytes] | StorageFile, *, content_type: str | None = None, metadata: dict[str, str] | None = None, overwrite: bool = False, ) -> str: """Save a file to the backend. Args: name: Path/key target. content: Raw byte content, file-like object, or async generator. content_type: MIME string (auto-detected if None). metadata: Custom tag key/values. overwrite: Replaces the file if True, otherwise appends an increments counter. Returns: The final saved relative path string. """ @abstractmethod async def open(self, name: str, mode: str = "rb") -> StorageFile: """Open a file handle for reading or writing. Returns: A StorageFile wrapper. """ @abstractmethod async def delete(self, name: str) -> None: """Deletes a file. Idempotent: does NOT raise if the file does not exist.""" @abstractmethod async def exists(self, name: str) -> bool: """Returns True if the file exists, False otherwise.""" @abstractmethod async def stat(self, name: str) -> StorageMetadata: """Returns metadata for the file. Raises FileNotFoundError if missing.""" @abstractmethod async def listdir(self, path: str = "") -> tuple[list[str], list[str]]: """List subdirectories and files in a path. Returns: A tuple of (directories_list, files_list). """ @abstractmethod async def size(self, name: str) -> int: """Returns the file size in bytes.""" @abstractmethod async def url(self, name: str, expire: int | None = None) -> str: """Generates a public URL or signed temporary URL. Args: expire: URL expiration period in seconds. """ StorageFile An asynchronous wrapper for reading and writing files. Implements the async context manager and iterator protocols. class StorageFile: @property def closed(self) -> bool: """Returns True if the file has been closed.""" async def read(self, size: int = -1) -> bytes: """Read up to size bytes. If -1, reads the entire file.""" async def write(self, data: bytes) -> int: """Write bytes to the file (requires a writable open mode).""" async def seek(self, offset: int, whence: int = 0) -> int: """Change the stream position relative to start (0), current (1), or end (2).""" async def tell(self) -> int: """Return the current stream position.""" async def close(self) -> None: """Release underlying system handles or HTTP connections.""" async def chunks(self, chunk_size: int = 65536) -> AsyncIterator[bytes]: """Stream the file in custom-sized byte chunks.""" StorageMetadata An immutable dataclass containing metadata for a stored file, returned by backend stat calls. @dataclass(frozen=True) class StorageMetadata: name: str # Relative path key size: int = 0 # Size in bytes content_type: str = "application/octet-stream" etag: str = "" # SHA-256 or remote MD5 hash last_modified: datetime | None = None created_at: datetime | None = None metadata: dict[str, str] = field(default_factory=dict) storage_class: str = "" # e.g., "STANDARD", "GLACIER" def to_dict(self) -> dict[str, Any]: """Convert metadata values to a serialized dictionary.""" Storage Fault Hierarchy Errors thrown by storage providers are normalized under the "storage" fault domain. Fault Class Fault Code Description ))} )
Go to Homepage