obstore下载 - obstore源码下载

obstore

其他源码

v0.2.0

下载

奥布斯托雷

与 Amazon S3、Google Cloud Storage、Azure Blob Storage 等对象存储服务以及 Cloudflare R2 等符合 S3 的 API 进行简单、快速的集成。

同步和异步 API。
具有可配置分块的流式下载。
流式list ，无需分页。
支持条件放置（“如果不存在则放置”）以及自定义标签和属性。
自动支持大型文件对象的分段上传。
可以选择以箭头形式返回列表结果，这比具体化 Python dict / list对象更快。
易于安装，无需任何 Python 依赖项。
底层 Rust 库具有生产质量并用于大规模生产系统，例如 Rust 包注册表 crates.io。
支持get_range和get_ranges中从 Rust 到 Python 的零拷贝数据交换。
具有静态类型检查的简单 API。
用于从环境变量和boto3.Session对象构建的帮助程序

安装

使用 pip 安装 obstore：

pip install obstore

Obstore 位于 conda-forge 上，可以使用 conda、mamba 或 pixi 安装。使用 conda 安装 obstore：

 conda install -c conda-forge obstore

文档

网站上提供了完整的文档。

用法

建造商店

构造商店的类从obstore.store子模块导出：

S3Store ：配置与 Amazon S3 的连接。
GCSStore ：配置与 Google Cloud Storage 的连接。
AzureStore ：配置与 Microsoft Azure Blob 存储的连接。
HTTPStore ：配置与通用 HTTP 服务器的连接
LocalStore ：提供相同对象存储接口的本地文件系统存储。
MemoryStore ：ObjectStore 的完全内存实现。

例子

 import boto3
from obstore . store import S3Store

session = boto3 . Session ()
store = S3Store . from_session ( session , "bucket-name" , config = { "AWS_REGION" : "us-east-1" })

配置

上面的每个存储类都有自己的配置，可以通过config命名参数访问。文档中对此进行了介绍，字符串文字位于类型提示中。

其他 HTTP 客户端配置可通过client_options命名参数获得。

与商店互动

所有与商店交互的方法都导出为顶级函数（而不是store对象上的方法）：

copy ：将对象从一个路径复制到同一对象存储中的另一个路径。
delete ：删除指定位置的对象。
get ：返回存储在指定位置的字节。
head ：返回指定位置的元数据
list ：列出具有给定前缀的所有对象。
put ：将提供的字节保存到指定位置
rename ：将同一对象存储中的对象从一个路径移动到另一个路径。

还有一些对于特定用例有用的附加 API：

get_range ：从文件中获取特定的字节范围。
get_ranges ：从单个文件中获取多个字节范围。
list_with_delimiter ：列出特定目录中的对象。
sign ：创建签名 URL。

所有方法都有一个类似的异步方法，具有相同的名称加上_async后缀。

例子

 import obstore as obs

store = obs . store . MemoryStore ()

obs . put ( store , "file.txt" , b"hello world!" )
response = obs . get ( store , "file.txt" )
response . meta
# {'path': 'file.txt',
#  'last_modified': datetime.datetime(2024, 10, 21, 16, 19, 45, 102620, tzinfo=datetime.timezone.utc),
#  'size': 12,
#  'e_tag': '0',
#  'version': None}
assert response . bytes () == b"hello world!"

byte_range = obs . get_range ( store , "file.txt" , offset = 0 , length = 5 )
assert byte_range == b"hello"

obs . copy ( store , "file.txt" , "other.txt" )
assert obs . get ( store , "other.txt" ). bytes () == b"hello world!"

所有这些方法也都有async对应方法，后缀为_async 。

 import obstore as obs

store = obs . store . MemoryStore ()

await obs . put_async ( store , "file.txt" , b"hello world!" )
response = await obs . get_async ( store , "file.txt" )
response . meta
# {'path': 'file.txt',
#  'last_modified': datetime.datetime(2024, 10, 21, 16, 20, 36, 477418, tzinfo=datetime.timezone.utc),
#  'size': 12,
#  'e_tag': '0',
#  'version': None}
assert await response . bytes_async () == b"hello world!"

byte_range = await obs . get_range_async ( store , "file.txt" , offset = 0 , length = 5 )
assert byte_range == b"hello"

await obs . copy_async ( store , "file.txt" , "other.txt" )
resp = await obs . get_async ( store , "other.txt" )
assert await resp . bytes_async () == b"hello world!"