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