使用pathlib管理S3路径。

Question

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

>>> from pathlib import Path

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

我该如何以这种方式操纵S3路径？

- beardc

---

就算不值一提，标准库的Path正在进行URI支持的努力，这将为支持各种URI打开一条道路。https://github.com/python/cpython/issues/107465 - undefined

Answer 1

使用s3path包

s3path包使得处理S3路径变得更加轻松。它可以从PyPI或conda-forge安装。使用S3Path类来处理实际的S3对象，否则使用PureS3Path，它不会实际访问S3。

虽然metaperture之前的回答提到了这个包，但没有包括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'

Windows用户可以使用.replace(os.sep, '/')来确保平台安全。

os.path.normpath不能被简单地使用：

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

- Asclepius

---

关于 os.path.join 的一点说明：“如果任何组件是绝对路径，则所有先前的路径组件都将被丢弃。”因此，您可能需要 .lstrip("/") 您的参数。 - ringo

os.path.join 在 Windows 上使用 "\"，所以这样做不起作用。 - Stefan

os.path.join在Windows上使用""，所以这样是行不通的。 - undefined

@Stefan，现在已经为Windows用户添加了一条注释。 - Asclepius

Answer 2

你可以尝试将urllib.parse与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))
# or
# s3_new_url = s3_url._replace(path=s3_path.as_posix()).geturl()
print(s3_new_url)

这有点麻烦，但这是你要求的。

- kichik

---

NotImplementedError: 无法在您的系统上实例化 'PosixPath'。Windows 用户应使用 PurePosixPath 替代。 - mediumnok

处理S3路径时，重要的是要注意它们使用不同于POSIX路径的转义方法。例如，如果您有一个带有空格的文件夹名称，在S3路径中应该使用"%20"而不是像在POSIX路径中使用反斜杠。由于这些差异，建议在处理S3路径时使用Urllib而不是Pathlib。 - Dmitry Balabka

处理S3路径时，重要的是要注意它们使用与POSIX路径不同的转义方法。例如，如果您的文件夹名称中有空格，您应该使用"%20"而不是像在POSIX路径中使用反斜杠。由于这些差异，建议在处理S3路径时使用Urllib而不是Pathlib。 - undefined

Answer 3

使用cloudpathlib

我想增加另一种选择，它具有良好的缓存和透明读/写访问，除了标准路径操作外。

cloudpathlib包提供对S3路径的pathlib方法支持，同时还支持Google Cloud Storage 和Azure Blob Storage。

例如：

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

- hume

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

- metaperture

Answer 5

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

- jwodder

---

这里有一次关于在某一天将URI功能添加到pathlib模块中的讨论。 - juanpa.arrivillaga

Answer 6

我同意@jwodder的答案，pathlib仅适用于文件系统路径。 不过，出于好奇心，我通过继承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安装依赖项。此外，您可以轻松地将其适应于任何其他URI，例如hdfs路径。

- zhukovgreen

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

- JizoSaves

Answer 8

将 str 类扩展以处理此类问题非常有用且简单。

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

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

- Alter