将 pathlib 用于 S3 路径

Question

我想构建一些功能以在 S3 和我的本地文件系统之间移动文件，但是 pathlib 似乎结合了重复的斜杠，破坏了我的 aws-cli 功能：

>>> from pathlib import Path

>>> str(Path('s3://loc'))
s3:/loc'

如何以这种方式操作 S3 路径？

Answer 1

没有。 pathlib 用于 文件系统 路径（即计算机上文件的路径），而 S3 路径是 URI。

Answer 2

您可以尝试组合 urllib.parse with pathlib.

from urllib.parse import urlparse, urlunparse
from pathlib import PosixPath

s3_url = urlparse('s3://bucket/key')
s3_path = PosixPath(s3_url.path)
s3_path /= 'hello'
s3_new_url = urlunparse((s3_url.scheme, s3_url.netloc, s3_path.as_posix(), s3_url.params, s3_url.query, s3_url.fragment))
print(s3_new_url)

挺麻烦的，不过是你求的。

Answer 3

使用 `s3path` 包

s3path package makes working with S3 paths a little less painful. It is installable from PyPI or conda-forge。对 S3 中的实际对象使用 S3Path class，否则使用实际上不应访问 S3 的 PureS3Path。

虽然确实提到了这个包，但它没有包括 URI 语法。

另请注意，此软件包存在某些缺陷，已在其问题中报告。

>>> from s3path import PureS3Path

>>> PureS3Path.from_uri('s3://mybucket/foo/bar') / 'add/me'
PureS3Path('/mybucket/foo/bar/add/me')

>>> _.as_uri()
's3://mybucket/foo/bar/add/me'

注意与 pathlib 的实例关系：

>>> from pathlib import Path, PurePath
>>> from s3path import S3Path, PureS3Path

>>> isinstance(S3Path('/my-bucket/some/prefix'), Path)
True
>>> isinstance(PureS3Path('/my-bucket/some/prefix'), PurePath)
True

使用`pathlib.Path`

这是 kichik 仅使用 pathlib 的答案的更懒惰版本。不一定推荐这种方法。只是并不总是完全有必要使用 urllib.parse.

>>> from pathlib import Path

>>> orig_s3_path = 's3://mybucket/foo/bar'
>>> orig_path = Path(*Path(orig_s3_path).parts[1:])
>>> orig_path
PosixPath('mybucket/foo/bar')

>>> new_path = orig_path / 'add/me'
>>> new_s3_path = 's3://' + str(new_path)
>>> new_s3_path
's3://mybucket/foo/bar/add/me'

使用`os.path.join`

仅用于简单联接，os.path.join如何？

>>> import os

>>> os.path.join('s3://mybucket/foo/bar', 'add/me')
's3://mybucket/foo/bar/add/me'
>>> os.path.join('s3://mybucket/foo/bar/', 'add/me')
's3://mybucket/foo/bar/add/me'

然而，

os.path.normpath 不能天真地使用：

>>> os.path.normpath('s3://mybucket/foo/bar')  # Converts 's3://' to 's3:/'
's3:/mybucket/foo/bar'

Answer 4

这是一个子类 pathlib.Path 用于 s3 路径的模块：https://pypi.org/project/s3path/

用法：

>>> from s3path import S3Path

>>> bucket_path = S3Path('/pypi-proxy/')
>>> [path for path in bucket_path.iterdir() if path.is_dir()]
[S3Path('/pypi-proxy/requests/'),
 S3Path('/pypi-proxy/boto3/'),
 S3Path('/pypi-proxy/botocore/')]

>>> boto3_package_path = S3Path('/pypi-proxy/boto3/boto3-1.4.1.tar.gz')
>>> boto3_package_path.exists()
True
>>> boto3_package_path.is_dir()
False
>>> boto3_package_path.is_file()
True

>>> botocore_index_path = S3Path('/pypi-proxy/botocore/index.html')
>>> with botocore_index_path.open() as f:
>>>     print(f.read())
"""
<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title>Package Index</title>
</head>
<body>
    <a href="botocore-1.4.93.tar.gz">botocore-1.4.93.tar.gz</a><br>
</body>
</html>
"""

Answer 5

扩展 str class 来处理这个问题既有用又简单

class URL(str):
  def __truediv__(self, val):
    return URL(self + '/' + val)

一个示例用法是 URL('s3://mybucket') / 'test' → "s3://mybucket/test"

Answer 6

使用`cloudpathlib`

希望将此添加为另一个选项，除了标准路径操作之外，它还具有良好的缓存和透明 read/write 访问权限。

除了 Google 云存储和 Azure Blob 存储之外，S3 路径的 cloupathlib package provides pathlib methods support。

例如：

from cloudpathlib import CloudPath
from itertools import islice

ladi = CloudPath("s3://ladi/Images/FEMA_CAP/2020/70349")

ladi.parent
#> S3Path('s3://ladi/Images/FEMA_CAP/2020')

ladi.bucket
#> 'ladi'

# list first 5 images for this incident
for p in islice(ladi.iterdir(), 5):
    print(p)
#> s3://ladi/Images/FEMA_CAP/2020/70349/DSC_0001_5a63d42e-27c6-448a-84f1-bfc632125b8e.jpg
#> s3://ladi/Images/FEMA_CAP/2020/70349/DSC_0002_a89f1b79-786f-4dac-9dcc-609fb1a977b1.jpg
#> s3://ladi/Images/FEMA_CAP/2020/70349/DSC_0003_02c30af6-911e-4e01-8c24-7644da2b8672.jpg
#> s3://ladi/Images/FEMA_CAP/2020/70349/DSC_0004_d37c02b9-01a8-4672-b06f-2690d70e5e6b.jpg
#> s3://ladi/Images/FEMA_CAP/2020/70349/DSC_0005_d05609ce-1c45-4de3-b0f1-401c2bb3412c.jpg

Answer 7

Pathy 非常适合这个： https://github.com/justindujardin/pathy

它使用引擎盖下的 Smart open 提供对存储桶存储的文件访问，因此比 s3path 更好。

您可以使用 Pathy.fluid 来制作适用于本地文件和存储桶中文件的 API

from pathlib import BasePath
from pathy import Pathy, FluidPath

def process_f(f: Union[Union[str, Pathy, BasePath]):
    path = Pathy.fluid(f)
    # now you have a Pathlib you can process that's local or in s3/GCS

Answer 8

我同意@jwodder 的回答，pathlib 仅适用于 fs 路径。然而，出于好奇，我对 pathlib.Path 的继承进行了一些尝试，并得到了一个非常可行的解决方案。

import pathlib


class S3Path(pathlib.PosixPath):
    s3_schema = "s3:/"

    def __new__(cls, *args, **kwargs):
        if args[0].startswith(cls.s3_schema):
            args = (args[0].replace(cls.s3_schema, "", 1),) + args[1:]
        return super().__new__(cls, *args, **kwargs)

    def __str__(self):
        try:
            return self.s3_schema + self._str
        except AttributeError:
            self._str = (
                self._format_parsed_parts(
                    self._drv,
                    self._root,
                    self._parts,
                )
                or "."
            )
            return self.s3_schema + self._str


def test_basic():
    s3_path_str: str = "s3://some/location"
    s3_path = S3Path(s3_path_str)
    assert str(s3_path) == s3_path_str

    s3_path_1 = s3_path / "here"
    assert str(s3_path_1) == s3_path_str + "/here"

    assert s3_path.parent == S3Path("s3://some")

它的优点是你不需要任何 pip install 依赖。另外，您可以轻松地将它应用于任何其他 URI，例如路径，例如 hdfs

将 pathlib 用于 S3 路径

Use pathlib for S3 paths

python

pathlib

使用 `s3path` 包

使用`pathlib.Path`

使用`os.path.join`

使用`cloudpathlib`

将 pathlib 用于 S3 路径

Use pathlib for S3 paths

python

pathlib

使用 s3path 包

使用pathlib.Path

使用os.path.join

使用cloudpathlib

使用 `s3path` 包

使用`pathlib.Path`

使用`os.path.join`

使用`cloudpathlib`