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!"