taolib.testing.file_storage.storage

taolib.testing.file_storage.storage#

存储后端模块。

导出存储协议和后端实现。

Submodules#

Classes#

`LocalStorageBackend`	本地文件系统存储后端实现。
`ObjectInfo`	存储对象信息。
`PartInfo`	分片上传部分信息。
`PutObjectResult`	对象写入结果。
`StorageBackendProtocol`	存储后端协议。
`S3StorageBackend`	S3 兼容存储后端实现。

Package Contents#

class taolib.testing.file_storage.storage.LocalStorageBackend(base_path: str = './storage')#

本地文件系统存储后端实现。

_base_path#

_multipart_uploads: dict[str, dict[str, Any]]#

_bucket_path(bucket: str) → pathlib.Path#

_object_path(bucket: str, key: str) → pathlib.Path#

async put_object(bucket: str, key: str, data: bytes | collections.abc.AsyncIterator[bytes], content_type: str = 'application/octet-stream', metadata: dict[str, str] | None = None) → taolib.testing.file_storage.storage.protocols.PutObjectResult#: 上传对象到本地文件系统。

async get_object(bucket: str, key: str) → collections.abc.AsyncIterator[bytes]#: 从本地文件系统读取对象。

async delete_object(bucket: str, key: str) → bool#: 从本地文件系统删除对象。

async copy_object(src_bucket: str, src_key: str, dst_bucket: str, dst_key: str) → taolib.testing.file_storage.storage.protocols.PutObjectResult#: 复制本地文件。

async head_object(bucket: str, key: str) → taolib.testing.file_storage.storage.protocols.ObjectInfo#: 获取本地文件元信息。

async object_exists(bucket: str, key: str) → bool#: 检查本地文件是否存在。

async list_objects(bucket: str, prefix: str = '', max_keys: int = 1000, continuation_token: str | None = None) → tuple[list[taolib.testing.file_storage.storage.protocols.ObjectInfo], str | None]#: 列出本地文件。

async generate_presigned_url(bucket: str, key: str, expires_in: int = 3600, method: str = 'GET') → str#: 生成本地文件伪签名 URL（仅用于开发）。

async create_multipart_upload(bucket: str, key: str, content_type: str = 'application/octet-stream') → str#: 创建本地分片上传会话。

async upload_part(bucket: str, key: str, upload_id: str, part_number: int, data: bytes) → taolib.testing.file_storage.storage.protocols.PartInfo#: 上传分片到本地临时目录。

async complete_multipart_upload(bucket: str, key: str, upload_id: str, parts: list[taolib.testing.file_storage.storage.protocols.PartInfo]) → taolib.testing.file_storage.storage.protocols.PutObjectResult#: 完成本地分片上传（合并文件）。

async abort_multipart_upload(bucket: str, key: str, upload_id: str) → None#: 中止本地分片上传。

async create_bucket(bucket: str) → None#: 创建本地存储桶目录。

async delete_bucket(bucket: str) → None#: 删除本地存储桶目录。

async bucket_exists(bucket: str) → bool#: 检查本地存储桶目录是否存在。

class taolib.testing.file_storage.storage.ObjectInfo#

存储对象信息。

key: str#

size: int#

last_modified: datetime.datetime#

etag: str#

content_type: str#

metadata: dict[str, str]#

class taolib.testing.file_storage.storage.PartInfo#

分片上传部分信息。

part_number: int#

etag: str#

class taolib.testing.file_storage.storage.PutObjectResult#

对象写入结果。

storage_path: str#

etag: str#

version_id: str | None = None#

class taolib.testing.file_storage.storage.StorageBackendProtocol#

Bases: Protocol

存储后端协议。

定义文件存储后端需要实现的所有操作。支持 S3 兼容存储、本地文件系统等多种后端。

async put_object(bucket: str, key: str, data: bytes | collections.abc.AsyncIterator[bytes], content_type: str = 'application/octet-stream', metadata: dict[str, str] | None = None) → PutObjectResult#: 上传对象。

async get_object(bucket: str, key: str) → collections.abc.AsyncIterator[bytes]#: 下载对象（流式）。

async delete_object(bucket: str, key: str) → bool#: 删除对象。

async copy_object(src_bucket: str, src_key: str, dst_bucket: str, dst_key: str) → PutObjectResult#: 复制对象。

async head_object(bucket: str, key: str) → ObjectInfo#: 获取对象元信息。

async object_exists(bucket: str, key: str) → bool#: 检查对象是否存在。

async list_objects(bucket: str, prefix: str = '', max_keys: int = 1000, continuation_token: str | None = None) → tuple[list[ObjectInfo], str | None]#

列出对象。

返回:: (对象列表, 下一页 token 或 None)

async generate_presigned_url(bucket: str, key: str, expires_in: int = 3600, method: str = 'GET') → str#: 生成预签名 URL。

async create_multipart_upload(bucket: str, key: str, content_type: str = 'application/octet-stream') → str#

创建分片上传会话。

返回:: upload_id

async upload_part(bucket: str, key: str, upload_id: str, part_number: int, data: bytes) → PartInfo#: 上传分片。

async complete_multipart_upload(bucket: str, key: str, upload_id: str, parts: list[PartInfo]) → PutObjectResult#: 完成分片上传。

async abort_multipart_upload(bucket: str, key: str, upload_id: str) → None#: 中止分片上传。

async create_bucket(bucket: str) → None#: 创建存储桶。

async delete_bucket(bucket: str) → None#: 删除存储桶。

async bucket_exists(bucket: str) → bool#: 检查存储桶是否存在。

class taolib.testing.file_storage.storage.S3StorageBackend(endpoint_url: str | None = None, access_key: str = '', secret_key: str = '', region: str = 'us-east-1')#

S3 兼容存储后端实现。

使用 aiobotocore 与 S3 兼容 API 交互。

_endpoint_url = None#

_access_key = ''#

_secret_key = ''#

_region = 'us-east-1'#

_session: Any = None#

_get_session() → Any#: 延迟初始化 aiobotocore session。

_create_client() → Any#: 创建 S3 客户端上下文管理器。

async put_object(bucket: str, key: str, data: bytes | collections.abc.AsyncIterator[bytes], content_type: str = 'application/octet-stream', metadata: dict[str, str] | None = None) → taolib.testing.file_storage.storage.protocols.PutObjectResult#: 上传对象到 S3。

async get_object(bucket: str, key: str) → collections.abc.AsyncIterator[bytes]#: 从 S3 下载对象（流式）。

async delete_object(bucket: str, key: str) → bool#: 从 S3 删除对象。

async copy_object(src_bucket: str, src_key: str, dst_bucket: str, dst_key: str) → taolib.testing.file_storage.storage.protocols.PutObjectResult#: 在 S3 上复制对象。

async head_object(bucket: str, key: str) → taolib.testing.file_storage.storage.protocols.ObjectInfo#: 获取 S3 对象元信息。

async object_exists(bucket: str, key: str) → bool#: 检查 S3 对象是否存在。

async list_objects(bucket: str, prefix: str = '', max_keys: int = 1000, continuation_token: str | None = None) → tuple[list[taolib.testing.file_storage.storage.protocols.ObjectInfo], str | None]#: 列出 S3 对象。

async generate_presigned_url(bucket: str, key: str, expires_in: int = 3600, method: str = 'GET') → str#: 生成 S3 预签名 URL。

async create_multipart_upload(bucket: str, key: str, content_type: str = 'application/octet-stream') → str#: 创建 S3 分片上传。

async upload_part(bucket: str, key: str, upload_id: str, part_number: int, data: bytes) → taolib.testing.file_storage.storage.protocols.PartInfo#: 上传 S3 分片。

async complete_multipart_upload(bucket: str, key: str, upload_id: str, parts: list[taolib.testing.file_storage.storage.protocols.PartInfo]) → taolib.testing.file_storage.storage.protocols.PutObjectResult#: 完成 S3 分片上传。

async abort_multipart_upload(bucket: str, key: str, upload_id: str) → None#: 中止 S3 分片上传。

async create_bucket(bucket: str) → None#: 创建 S3 存储桶。

async delete_bucket(bucket: str) → None#: 删除 S3 存储桶。

async bucket_exists(bucket: str) → bool#: 检查 S3 存储桶是否存在。

taolib.testing.file_storage.storage

目录

taolib.testing.file_storage.storage#

Submodules#

Classes#

Package Contents#