TL;DR
PythonでS3のファイルパスを扱う場合には s3pathlib が便利です。組み込みモジュールの pathlib では s3:// から始まるS3のパスをパースすることができませんが s3pathlib なら pathlib と同じように扱うことができます。
from pathlib import Path
from s3pathlib import S3Path
local_dir = Path("./my_dir")
s3_dir = S3Path("s3://my_bucket/my_dir")
if local_dir.exists():
print(f"ローカルのディレクトリ {local_dir} は既に存在します。")
if s3_dir.exists():
print(f"S3上のディレクトリ {s3_dir} は既に存在します。")
私は s3pathlib を知る前は str 型でS3上のパスを処理していたのですが、 str 型でパスを記述することは型の安全性の観点から推奨されませんし、上記の exists() や joinpath() , parent といった pathlib で実装されている機能を代替するのに手間がかかることからフラストレーションを感じていました。
そんな悩みも s3pathlib で解決です。
[https://s3pathlib.readthedocs.io/en/latest/index.html:embed:cite]
！ S3は厳密にはディレクトリ構造を持たずKey/Valueによるフラットな構造を採用していますが、疑似的にディレクトリ構造のように扱うことができます。また s3pathlib のドキュメント上でもディレクトリという用語が使用されています。そのため、この記事でも便宜上ディレクトリという用語を使用します。
Amazon S3 has a flat structure instead of a hierarchy like you would see in a file system.
However, for the sake of organizational simplicity, the Amazon S3 console supports the folder concept as a means of grouping objects.
[https://docs.aws.amazon.com/AmazonS3/latest/userguide/using-folders.html:embed:cite]
環境
AWS CLIが使用できる環境をご利用ください。
インストール方法
pipで配布されています。
pip install s3pathlib
s3pathlib チートシート
よく使う機能を一通りまとめます。
S3Path オブジェクトを定義する
ファイルの場合
- パスをそのまま貼り付ける
s3_file = S3Path("s3://my_bucket/my_dir/my_file.txt")
- パスをバケット・ディレクトリ・ファイルに分割して指定する
s3_file = S3Path("my_bucket", "my_dir", "my_file.txt")
- パスをARN表記で指定する
s3_file = S3Path("arn:aws:s3:::my_bucket/my_dir/my_file.txt")
ディレクトリの場合
ファイルの場合とほとんど同じ。
- パスをそのまま貼り付ける
s3_dir = S3Path("s3://my_bucket/my_dir/")
- パスをバケット・ディレクトリ・ファイルに分割して指定する
s3_dir = S3Path("my_bucket", "my_dir")
- パスをARN表記で指定する
s3_dir = S3Path("arn:aws:s3:::my_bucket/my_dir/)
-
！ 末尾が
/でない場合はディレクトリとして認識されないので注意する
>>> s3_dir = S3Path("s3://my_bucket/my_dir/")
>>> print(s3_dir.is_dir())
True
>>> s3_dir = S3Path("s3://my_bucket/my_dir")
>>> print(s3_dir.is_dir())
False
S3Path オブジェクトの情報を取得する
！ 便宜上 S3Path のメンバとして表記していますが、ソースはMixInクラス上にあります。記事中の他のメンバについても同様です。
S3Path.uri: URIを取得する
>>> print(s3_file.uri)
s3://my_bucket/my_dir/my_file.txt
S3Path.bucket: バケットを取得する
>>> print(s3_file.bucket)
my_bucket/my_dir/my_file.txt
S3Path.key: パスを取得する
>>> print(s3_file.key)
my_bucket/my_dir/my_file.txt
アップロード
S3path.upload_file(): ファイルをアップロード
- ローカル上に存在する
my_file.txtをS3のパスs3://my_bucket/my_dir/my_file.txtへアップロード
local_file = Path("my_file.txt")
s3_file = S3Path("s3://my_bucket/my_dir/my_file.txt")
s3_file.upload_file(local_file)
- 上書きを許可する場合
local_file = Path("my_file.txt")
s3_file = S3Path("s3://my_bucket/my_dir/my_file.txt")
s3_file.upload_file(local_file, overwrite=True)
S3path.upload_dir(): ディレクトリをアップロード
- ローカル上に存在する
my_dir/をS3のパスs3://my_bucket/my_dir/へアップロード
local_dir = Path("my_dir/")
s3_dir = S3Path("s3://my_bucket/my_dir/")
s3_dir.upload_dir(local_dir)
- 上書きを許可する場合
local_dir = Path("my_dir/")
s3_dir = S3Path("s3://my_bucket/my_dir/")
s3_dir.upload_dir(local_dir, overwrite=True)
ダウンロード
boto3 モジュールに S3Path を組み合わせることで、ダウンロード用のスクリプトを記述できます。
ファイルをダウンロード
単一のファイルをダウンロードする機能は、次のような例で実装できます。
from pathlib import Path
import boto3
from s3pathlib import S3Path
def download_file_from_s3(src_path: S3Path, *, dst_dir: Path = Path("./")):`
"""Download a file from input S3 path.
Args:
src_path (S3Path): Directory of file path on S3.
dst_dir (Path, optional): Directory path on local.
"""
s3_resource = boto3.resource("s3")
bucket = s3_resource.Bucket(src_path.bucket)
dst_path = dst_dir.joinpath(Path(src_path).name)
bucket.download_file(src_path, dst_path)
print(f"Downloaded to {dst_path}")
ディレクトリをダウンロード
また、ディレクトリ毎すべてのファイルとサブディレクトリをダウンロードする機能は、次のように実装できます。
from pathlib import Path
import boto3
from s3pathlib import S3Path
from tqdm import tqdm
def download_from_s3(src_path: S3Path, *, dst_dir: Path = Path("./")):
"""Download all files from input S3 path.
Args:
src_path (S3Path): Directory of file path on S3.
dst_dir (Path, optional): Directory path on local.
"""
s3_resource = boto3.resource("s3")
bucket = s3_resource.Bucket(src_path.bucket)
dst_path = Path("./")
for filepath in tqdm(get_s3_files(src_path)):
dst_path = dst_dir.joinpath(Path(filepath).name)
dst_path.parent.mkdir(exist_ok=True, parents=True)
if dst_path.exists():
print(f"{dst_path} already exists.")
continue
bucket.download_file(filepath, dst_path)
print(f"Downloaded to {dst_path}")
その他S3の操作
S3Path.mkdir(): S3上にディレクトリを作成する
>>> s3_dir = S3Path("s3://my_bucket/my_dir/").to_dir()
>>> s3_dir.mkdir(exist_ok=True)
S3Path.write_text(), S3Path.read_text(): S3上のファイルに読み書きする
>>> s3_file = S3Path("s3://my_bucket/my_dir/my_file.txt")
>>> s3_file.write_text("hoge")
>>> s3_file.read_text()
'hoge'
その他pathlibでよく使う機能に相当するもの
S3Path.joinpath(): 下の階層のパスをつなげる
>>> s3_dir = S3Path("s3://my_bucket/my_dir/")
>>> s3_file = s3_dir.joinpath("my_file.txt")
>>> print(s3_file.uri)
s3://my_bucket/my_dir/my_file.txt
S3Path.exists(): S3上に存在するか確かめる
- ディレクトリでもファイルでも同様
>>> s3_dir = S3Path("s3://my_bucket/my_dir/")
>>> print(s3_dir.exists()
True
>>> s3_dir = S3Path("s3://my_bucket/pseudo_my_dir/")
>>> print(s3_dir.exists())
False
S3Path.is_dir(): パスがディレクトリかファイルか判定する
>>> s3_dir = S3Path("s3://my_bucket/my_dir/")
>>> print(s3_dir.is_dir())
True
まとめ
一通り s3pathlib の主要な機能を挙げてみましたが、この中でも特に joinpath() とアップロード・ダウンロードの組み合わせは大変便利です。これだけでS3のパスを str で記述して split() でバラして...というような非効率なコードから解放されるようになります。
S3を頻繁に使うPythonエンジニアのみなさんはぜひ使ってみてください。
本記事は下記の記事と同じ内容です。 アクセス解析を目的としてマルチポストしています。
https://tech-blog.yayoi-kk.co.jp/entry/2024/07/02/110000
また弥生では一緒に働く仲間を募集しています。 ぜひエントリーお待ちしております。
https://herp.careers/v1/yayoi/requisition-groups/2363876d-7695-46ea-9e6a-0bc64d11b6bc