Pathy：用于本地和云存储桶存储的Path接口

Pathy是一个python包（带有类型注解），用于使用pathlib接口处理云存储桶存储提供商。它提供了一个易于使用的API，附带一个CLI应用程序，用于在本地文件和远程存储桶之间进行基本文件操作。它通过允许开发者在开发过程中针对本地文件系统进行操作，仅在部署时切换到实时API，从而实现流畅的开发体验。它还通过可选的本地文件缓存，使将存储桶块转换为本地文件变得简单。

🚀 快速入门

您可以从pip安装pathy

pip install pathy

该包导出Pathy类和配置存储桶存储提供商的实用工具。

from pathy import Pathy, use_fs
# Use the local file-system for quicker development
use_fs()
# Create a bucket
Pathy("gs://my_bucket").mkdir(exist_ok=True)
# An excellent blob
greeting = Pathy(f"gs://my_bucket/greeting.txt")
# But it doesn't exist yet
assert not greeting.exists()
# Create it by writing some text
greeting.write_text("Hello World!")
# Now it exists
assert greeting.exists()
# Delete it
greeting.unlink()
# Now it doesn't
assert not greeting.exists()

支持的云服务

下表详细说明了支持的云提供商API。

Google Cloud Storage

Google建议使用JSON凭证文件，您可以通过路径指定它

from google.oauth2 import service_account
from pathy import set_client_params

credentials = service_account.Credentials.from_service_account_file("./my-creds.json")
set_client_params("gs", credentials=credentials)

Amazon S3

S3使用JSON凭证文件，您可以通过路径指定它

from pathy import set_client_params

set_client_params("s3", key_id="YOUR_ACCESS_KEY_ID", key_secret="YOUR_ACCESS_SECRET")

Azure

Azure blob存储可以传递一个connection_string

from pathy import set_client_params

set_client_params("azure", connection_string="YOUR_CONNECTION_STRING")

或一个BlobServiceClient实例

from azure.storage.blob import BlobServiceClient
from pathy import set_client_params

service: BlobServiceClient = BlobServiceClient.from_connection_string(
    "YOUR_CONNECTION_STRING"
)
set_client_params("azure", service=service)

语义版本控制

在Pathy达到v1.0之前，该项目不保证API保持一致，这意味着类型和类可能会移动或被删除。话虽如此，我们在处理破坏性更改时尽量保持可预测性，因此该项目使用语义版本控制来帮助用户避免破坏。

具体来说，新版本通过增加patch（修补）semver组件来引入新功能和修复，当有重大变更时增加minor（次要）组件。如果您不太了解semver字符串，它们通常格式为{major}.{minor}.{patch}，增加patch组件意味着增加最后一个数字。

考虑一些例子

如果您担心破坏性变更，可以在要求中固定版本，以便它不超过当前的semver minor组件，例如如果当前版本是0.1.37

pathy>=0.1.37,<0.2.0

🎛 API

Pathy 类

Pathy(self, args, kwargs)

是pathlib.Path的子类，适用于与存储桶API一起使用。

exists 方法

Pathy.exists(self) -> bool

如果路径指向一个现有的存储桶、块或前缀，则返回True。

fluid 类方法

Pathy.fluid(
    path_candidate: Union[str, Pathy, BasePath],
) -> Union[Pathy, BasePath]

从输入路径或字符串推断Pathy或pathlib.Path。

返回类型是潜在FluidPath类型的联合体，并且将正确地对所有输入类型的最低重叠API进行类型检查。

如果您需要使用类型的特定实现细节，可以将此函数的返回“缩小”到所需类型，例如

from pathy import FluidPath, Pathy

fluid_path: FluidPath = Pathy.fluid("gs://my_bucket/foo.txt")
# Narrow the type to a specific class
assert isinstance(fluid_path, Pathy), "must be Pathy"
# Use a member specific to that class
assert fluid_path.prefix == "foo.txt/"

from_bucket 类方法

Pathy.from_bucket(bucket_name: str, scheme: str = 'gs') -> 'Pathy'

从存储桶名称初始化Pathy。此辅助函数添加尾随斜杠和适当的前缀。

from pathy import Pathy

assert str(Pathy.from_bucket("one")) == "gs://one/"
assert str(Pathy.from_bucket("two")) == "gs://two/"

glob 方法

Pathy.glob(
    self: 'Pathy',
    pattern: str,
) -> Generator[Pathy, NoneType, NoneType]

在Pathy实例相对路径上执行glob匹配，返回所有匹配的块。

is_dir 方法

Pathy.is_dir(self: 'Pathy') -> bool

确定路径是否指向存储桶或存储桶中给定块的前缀。

如果路径指向存储桶或块前缀，则返回True。如果路径指向块或路径不存在，则返回False。

is_file 方法

Pathy.is_file(self: 'Pathy') -> bool

确定路径是否指向存储桶中的块。

如果路径指向块，则返回True。如果路径指向存储桶或块前缀，或路径不存在，则返回False。

iterdir 方法

Pathy.iterdir(
    self: 'Pathy',
) -> Generator[Pathy, NoneType, NoneType]

遍历在给定的存储桶或块前缀路径中找到的块。

ls 方法

Pathy.ls(self: 'Pathy') -> Generator[BlobStat, NoneType, NoneType]

列出给定路径下的块名称及其状态信息。

如果您还需要枚举块的统计信息，这比使用iterdir要快得多。

为每个找到的块产生BlobStat对象。

mkdir 方法

Pathy.mkdir(
    self,
    mode: int = 511,
    parents: bool = False,
    exist_ok: bool = False,
) -> None

从给定路径创建存储桶。由于存储桶API只有隐式文件夹结构（由重叠前缀的块的存在来决定），这不会做任何事情，只是创建存储桶。

如果parents为False，只有当路径指向存储桶且不指向其他内容时，才会创建存储桶。如果parents为True，即使路径指向特定的块，也会创建存储桶。

忽略mode参数。

如果exist_ok为False且存储桶已存在，则引发FileExistsError。

open 方法

Pathy.open(
    self: 'Pathy',
    mode: str = 'r',
    buffering: int = 8192,
    encoding: Optional[str] = None,
    errors: Optional[str] = None,
    newline: Optional[str] = None,
) -> IO[Any]

以流式传输方式打开给定的块。这委托给smart_open库，该库处理多个存储桶API提供商的大型文件流。

owner 方法

Pathy.owner(self: 'Pathy') -> Optional[str]

返回指向存储桶或块的所有者的用户名。如果所有者未知或不支持存储桶API提供商，则返回None。

rename 方法

Pathy.rename(self: 'Pathy', target: Union[str, pathlib.PurePath]) -> 'Pathy'

将此路径重命名为指定的目标。

如果目标存在且是文件，则如果用户有权限，将静默替换它。

如果路径是块前缀，则将替换所有具有相同前缀的块以匹配目标前缀。

replace 方法

Pathy.replace(self: 'Pathy', target: Union[str, pathlib.PurePath]) -> 'Pathy'

将此路径重命名为指定的目标。

如果目标指向现有的路径，则将其替换。

resolve 方法

Pathy.resolve(self, strict: bool = False) -> 'Pathy'

解析给定路径以移除任何相对路径指定符。

from pathy import Pathy

path = Pathy("gs://my_bucket/folder/../blob")
assert path.resolve() == Pathy("gs://my_bucket/blob")

rglob 方法

Pathy.rglob(
    self: 'Pathy',
    pattern: str,
) -> Generator[Pathy, NoneType, NoneType]

相对于此Pathy实例执行递归glob匹配，返回所有匹配的blob。想象在调用glob之前添加"**/"。

rmdir 方法

Pathy.rmdir(self: 'Pathy') -> None

删除此存储桶或blob前缀。它必须是空的。

samefile 方法

Pathy.samefile(
    self: 'Pathy',
    other_path: Union[str, bytes, int, pathlib.Path],
) -> bool

确定此路径是否指向其他_path指向的同一位置。

stat 方法

Pathy.stat(self: 'Pathy') -> pathy.BlobStat

返回关于此存储桶路径的信息。

to_local 类方法

Pathy.to_local(
    blob_path: Union[Pathy, str],
    recurse: bool = True,
) -> pathlib.Path

下载并缓存一个blob或一组匹配前缀的blob。

缓存对文件更新时间敏感，当其更新时间戳更改时下载新的blob。

touch 方法

Pathy.touch(self: 'Pathy', mode: int = 438, exist_ok: bool = True) -> None

在此路径创建一个blob。

如果blob已存在，当exist_ok为true时函数成功（并且其修改时间更新为当前时间），否则引发FileExistsError。

BlobStat 数据类

BlobStat(
    self,
    name: str,
    size: Optional[int],
    last_modified: Optional[int],
) -> None

存储桶项的stat

use_fs 函数

use_fs(
    root: Optional[str, pathlib.Path, bool] = None,
) -> Optional[pathy.BucketClientFS]

使用本地文件系统的路径来存储blob和存储桶。

这在开发和测试环境中以及嵌入式应用程序中很有用。

get_fs_client 函数

get_fs_client() -> Optional[pathy.BucketClientFS]

获取文件系统客户端（或None）

use_fs_cache 函数

use_fs_cache(
    root: Optional[str, pathlib.Path, bool] = None,
) -> Optional[pathlib.Path]

使用本地文件系统的路径来缓存blob和存储桶。

这在您想避免多次获取大型blob或需要将本地文件路径传递给第三方库时很有用。

get_fs_cache 函数

get_fs_cache() -> Optional[pathlib.Path]

获取包含文件系统缓存的blob和时间的文件夹。

set_client_params 函数

set_client_params(scheme: str, kwargs: Any) -> None

指定实例化特定于服务的Client对象时传递的参数。这允许以底层客户端库所偏好的方式传递凭据。

CLI

Pathy命令行界面。（v0.5.2）

用法:

$ [OPTIONS] COMMAND [ARGS]...

选项:

• --install-completion：为当前shell安装补全。
• --show-completion：显示当前shell的补全，以便复制或自定义安装。
• --help：显示此消息并退出。

命令:

• cp：从一处复制blob或blob文件夹到另一处...
• ls：列出给定位置存在的blob。
• mv：将blob或blob文件夹从一个路径移动到另一个路径...
• rm：从给定位置删除blob或blob文件夹。

cp

从一个存储桶复制blob或blob文件夹到另一个存储桶。

用法:

$ cp [OPTIONS] FROM_LOCATION TO_LOCATION

参数:

• FROM_LOCATION：[必需]
• TO_LOCATION：[必需]

选项:

• --help：显示此消息并退出。

ls

列出给定位置存在的blob。

用法:

$ ls [OPTIONS] LOCATION

参数:

• LOCATION：[必需]

选项:

• -l, --long：打印带有更新时间和大小的长格式条目。[默认：False]
• --help：显示此消息并退出。

mv

将blob或blob文件夹从一个路径移动到另一个路径。

用法:

$ mv [OPTIONS] FROM_LOCATION TO_LOCATION

参数:

• FROM_LOCATION：[必需]
• TO_LOCATION：[必需]

选项:

• --help：显示此消息并退出。

rm

从给定位置删除blob或blob文件夹。

用法:

$ rm [OPTIONS] LOCATION

参数:

• LOCATION：[必需]

选项:

• -r, --recursive：递归删除文件和文件夹。[默认：False]
• -v, --verbose：打印删除的文件和文件夹。[默认：False]
• --help：显示此消息并退出。

致谢

Pathy最初基于S3Path项目，该项目为S3存储桶提供路径接口。