Zenn
Open5

Apache OpenDALをPythonから利用する.

Python
S3
Cloudflare
file
R2
shouseishousei

そもそもOpenDALとは

Open Data Access Layer: Access data freely
要は

ファイルの読み書きを抽象化して、いろんなクラウドサービスやネットワークを介した読み書きも対応できるようにしましたよ. なやつ.

ホームページ
CoreはRustで書かれている. Python binding, Node.js bindiingあり.

対応している形式一覧.ここ参照

📄️ Azblob
Azure Storage Blob services support.

📄️ Azdfs
Azure Data Lake Storage Gen2 Support.

📄️ COS
Huawei Cloud COS services support.

📄️ DashMap
dashmap backend support.

📄️ Fs
POSIX file system support.

📄️ FTP
FTP and FTPS services support.

📄️ HDFS
Hadoop Distributed File System (HDFS™) support.

📄️ IPFS
IPFS file system support based on IPFS HTTP Gateway.

📄️ IPMFS
IPFS Mutable File System (IPMFS) support

📄️ Memory
In memory service support. (BTreeMap Based)

📄️ OSS
Aliyun Object Storage Service (OSS) support.

📄️ Redis
Redis services support.

📄️ RocksDB
RocksDB service support.

📄️ S3
Aws S3 and compatible services (including minio, digitalocean space, Tencent Cloud Object Storage(COS) and so on) support.

📄️ SFTP
SFTP services support. (only works on unix)

📄️ Sled
Sled service support.

📄️ WebHDFS
WebHDFS's REST API support.
shouseishousei

試しにCloudflareのR2書き込みができるか試してみる.

ドキュメンテーション読んでもS3 compatibleなサービスとの接続方法の詳細書いてない.
Rustからの使い方なら書いてあるのでこれを参考にしてみる.
RustのようなBuilder patternのAPIは用意されていないようなので、試しに読み書き操作するための接続を確立する??クラス Operator/AsyncOperatorのコンストラクタで操作できるか試す.

以下のやり方でいけた.

from opendal import AsyncOperator
from pathlib import Path
import asyncio

# 接続に必要な事前の情報 (R2のページに行って取ってくる)
R2_ENDPOINT = "https://xxxx"
R2_AWS_ACCESS_KEY_ID = "xxxyyy"
R2_AWS_ACCESS_KEY_SECRET = "xxxyyyzzz"
bucket_name = "my-bucket"  # 予め作ったBucket名

async_op = AsyncOperator("s3",   # 第一引数 schemeにはサービス名を指定, "s3"と指定する.
                        # これ以降の引数は**kwargsとしか書かれていない. 多分RustのBuilderで指定されるメソッド名をそのまま利用するだけで行けるのだと思われる.
                         bucket=bucket_name,
                         root="/tmp",  # {bucket_name}/tmp/{my_path}のように書き込まれる. 絶対パスで"/"から記述しないといけないらしい.
                         endpoint=R2_ENDPOINT,
                         access_key_id=R2_AWS_ACCESS_KEY_ID,
                         secret_access_key=R2_AWS_ACCESS_KEY_SECRET,
                         region="auto"  # cloudflare r2は regionは auto
                         )

async def write(aop: AsyncOperator, src_path: Path):
    # ローカルのファイルデータを`Path`を界して /{bucket_name}/tmp/{src_path.name} という名前でr2に書き込む
    await aop.write(src_path.name, src_path.read_bytes())

if __name__ == "__main__":
    asyncio.run(write(async_op,  Path("src_image.jpeg"))
shouseishousei

Operator クラスもあったがこちらはS3は対応してないらしくやろうとするとエラーが起こる.

shouseishousei

R2から読み込み -> ローカルに書き込み

async def read(aop: AsyncOperator, name: str) -> bytes:
    return await aop.read(name)

# localにダウンロードする例
async def download(aop: AsyncOperator,src_name: str, local_dst: Path):
    b = read(aop, src_name)
    local_dst.write_bytes(b)
    

if __name__ == "__main__":
    asyncio.run(download(async_op, src_name="src_image.jpeg", dst_name = Path("src_image.jpeg"))

読み込み先名前指定はは先ほどの書き込み先で指定した名前と同じアクセス方法で可能