from __future__ import annotations from abc import ABC, abstractmethod from typing import BinaryIO, Dict, Optional class StorageInterface(ABC): """存储接口抽象基类""" storage_type: str @abstractmethod def upload(self, file_stream: BinaryIO, key: str, mime_type: Optional[str] = None) -> Dict: """ 上传文件 Args: file_stream: 文件流 key: 存储路径/键名 mime_type: MIME类型 Returns: 存储元信息字典(如ETag、VersionId等) """ pass @abstractmethod def append_chunk(self, key: str, chunk_stream: BinaryIO, offset: int) -> None: """ 追加写入数据块(用于TUS协议) Args: key: 存储路径 chunk_stream: 数据块流 offset: 写入偏移量 """ pass @abstractmethod def download(self, key: str) -> BinaryIO: """ 下载文件 Args: key: 存储路径 Returns: 文件流 """ pass @abstractmethod def delete(self, key: str) -> None: """ 删除文件 Args: key: 存储路径 """ pass @abstractmethod def exists(self, key: str) -> bool: """ 检查文件是否存在 Args: key: 存储路径 Returns: 是否存在 """ pass @abstractmethod def get_url(self, key: str, expires: int = 3600) -> str: """ 获取文件访问URL Args: key: 存储路径 expires: 过期时间(秒),0表示永久 Returns: 访问URL """ pass # 以下为可选扩展:预览与缩略图URL(默认实现回退到 get_url) def get_preview_url(self, key: str, expires: int = 3600) -> str: """获取预览URL,默认与 get_url 相同(子类可覆盖提供更优预览能力)""" return self.get_url(key, expires) def get_thumbnail_url( self, key: str, width: int = 200, height: int = 200, mode: str = "fit", expires: int = 3600, ) -> Optional[str]: """获取缩略图URL,默认返回原URL,子类可覆盖使用存储侧图片处理能力。 Args: key: 对象键 width: 目标宽度 height: 目标高度 mode: fit/fill/pad 等模式 expires: 过期秒数 """ return self.get_url(key, expires)