地位

兼容性

这个包不再为 python 3.6 维护。可用于 python 3.6 的最新版本是 0.1.129。请使用 python >= 3.7。

接触

透明路径

一个类，它允许人们以与使用pathlib.Path对象相同的方式使用本地文件系统或 Google Cloud Storage (GCS) 文件系统中的路径。可以一次使用多个不同的 GCP 项目。

要求

您将需要 GCP 凭据，可以是.json文件，可以在 envvar GOOGLE_APPLICATION_CREDENTIALS 中设置，也可以直接在谷歌云实例（VM、pod 等）中运行。

安装

您可以使用 pip 安装此软件包：

pip install transparentpath

可选包

香草版本允许您声明路径并使用它们。您可以在内置open方法中使用它们。或者，您还可以安装对其他几个包的支持，如 pandas、dask 等……当前可用的可选包可通过以下命令访问：

pip install transparentpath[pandas]
pip install transparentpath[parquet]
pip install transparentpath[hdf5]
pip install transparentpath[json]
pip install transparentpath[excel]
pip install transparentpath[dask]

您可以一次安装所有这些

pip install transparentpath[all]

用法

创建指向 GCS 的路径和不指向 GCS 的路径：

from transparentpath import Path
# Or : from transparentpath import TransparentPath
p = Path("gs://mybucket/some_path", token="some/cred/file.json")
p2 = p / "foo"  # Will point to gs://mybucket/some_path/foo
p3 = Path("bar")  # Will point to local path "bar"

将所有路径设置为默认指向 GCS：

from transparentpath import Path
Path.set_global_fs("gcs", token="some/cred/file.json")
p = Path("mybucket") / "some_path" # Will point to gs://mybucket/some_path
p2 = p / "foo"  # Will point to gs://mybucket/some_path/foo
p3 = Path("bar", fs="local")  # Will point to local path "bar"
p4 = Path("other_bucket")  # Will point to gs://other_bucket (assuming other_bucket is a bucket on GCS)
p5 = Path("not_a_bucket")  # Will point to local path "not_a_bucket" (assuming it is indeed, not a bucket on GCS)

默认情况下，将所有路径设置为指向多个 GCS 项目：

from transparentpath import Path
Path.set_global_fs("gcs", token="some/cred/file.json")
Path.set_global_fs("gcs", token="some/other/cred/file.json")
p = Path("mybucket") / "some_path" # Will point to gs://mybucket/some_path
p2 = p / "foo"  # Will point to gs://mybucket/some_path/foo
p3 = Path("bar", fs="local")  # Will point to local path "bar"
p4 = Path("other_bucket")  # Will point to gs://other_bucket (assuming other_bucket is a bucket on GCS)
p5 = Path("not_a_bucket")  # Will point to local path "not_a_bucket" (assuming it is indeed, not a bucket on GCS)

这里，mybucket和other_bucket可以在两个不同的项目上，只要至少有一个凭证文件可以访问它们。

设置所有路径默认指向 GCS，并指定默认存储桶：

from transparentpath import Path
Path.set_global_fs("gcs", bucket="mybucket", token="some/cred/file.json")
p = Path("some_path")  # Will point to gs://mybucket/some_path/
p2 = p / "foo"  # Will point to gs://mybucket/some_path/foo
p3 = Path("bar", fs="local")  # Will point to local path "bar"
p4 = Path("other_bucket")  # Will point to gs://mybucket/other_bucket
p5 = Path("not_a_bucket")  # Will point to gs://mybucket/not_a_bucket

如果您的代码应该能够以有时是远程的，有时是本地的路径运行，那么最新的选项很有趣。为此，您可以使用 class 属性nas_dir。然后当创建路径时，如果它以nas_dir' 路径开头， 则nas_dir' 路径将替换为存储桶名称。例如，如果您在本地备份存储桶，例如/my/local/backup ，这将非常有用。然后你可以这样做：

from transparentpath import Path
Path.nas_dir = "/my/local/backup"
Path.set_global_fs("gcs", bucket="mybucket", token="some/cred/file.json")
p = Path("some_path")  # Will point to gs://mybucket/some_path/
p3 = Path("/my/local/backup") / "some_path"  # Will ALSO point to gs://mybucket/some_path/

from transparentpath import Path
Path.nas_dir = "/my/local/backup"
# Path.set_global_fs("gcs", bucket="mybucket", token="some/cred/file.json")
p = Path("some_path")  # Will point to /my/local/backup/some_path/
p3 = Path("/my/local/backup") / "some_path"  # Will ALSO point to /my/local/backup/some_path/

在前面的所有示例中，token如果设置了环境变量 GOOGLE_APPLICATION_CREDENTIALS并指向.json凭证文件，或者如果您的代码在可以访问 GCS 的 GCP 机器（VM、集群...）上运行，则可以省略该参数。

无论您是使用 GCS 还是本地文件系统，以下是 TransparentPath 可以做什么的示例：

from transparentpath import Path
# Path.set_global_fs("gcs", bucket="bucket_name", project="project_name")
# The following lines will also work with the previous line uncommented 

# Reading a csv into a pandas' DataFrame and saving it as a parquet file
mypath = Path("foo") / "bar.csv"
df = mypath.read(index_col=0, parse_dates=True)
otherpath = mypath.with_suffix(".parquet")
otherpath.write(df)

# Reading and writing a HDF5 file works on GCS and on local:
import numpy as np
mypath = Path("foo") / "bar.hdf5"  # can be .h5 too
with mypath.read() as ifile:
    arr = np.array(ifile["store1"])

# Doing '..' from 'foo/bar.hdf5' will return 'foo'
# Then doing 'foo' + 'babar.hdf5' will return 'foo/babar.hdf5' ('+' and '/' are synonymes)
mypath.cd("..")  # Does not return a path but modifies inplace
with (mypath  + "babar.hdf5").write(None) as ofile:
    # Note here that we must explicitely give 'None' to the 'write' method in order for it
    # to return the open HDF5 file. We could also give a dict of {arr: "store1"} to directly
    # write the file.
    ofile["store1"] = arr


# Reading a text file. Can also use 'w', 'a', etc... also works with binaries.
mypath = Path("foo") / "bar.txt"
with open(mypath, "r") as ifile:
    lines = ifile.readlines()

# open is overriden to understand gs://
with open("gs://bucket/file.txt", "r") as ifile:
    lines = ifile.readlines()

mypath.is_file()
mypath.is_dir()
mypath.is_file()
files = mypath.parent.glob("*.csv")  # Returns a Iterator[TransparentPath], can be casted to list

从前面的示例中可以看出，所有从透明路径返回路径的方法都返回一个透明路径。

达斯克

TransparentPath 支持本地和远程从 csv、excel、parquet 和 HDF5 写入和读取 Dask 数据帧。您需要安装 dask-dataframe 和 dask-distributed，如果您运行pip install transparentpath[dask]. 编写 Dask 数据帧不需要传递任何附加参数，因为在调用适当的写入方法之前将检查类型。但是，阅读需要您将use_dask参数传递给该read()方法。如果要读取的文件是 HDF5，您还需要指定 set_names，匹配Dask方法的参数键。read_hdf()

请注意，如果读取远程 HDF5，文件将下载到本地 tmp 中，然后读取。如果不使用 Dask，则文件在读取后会被删除。但由于 Dask 使用延迟进程，可能会在实际读取文件之前删除文件，因此会保留文件。如果系统未自动完成 /tmp 目录，则由您自行清空。

行为

即使使用相对路径创建，TransparentPath 的所有实例都是绝对的。

TransparentPaths 被视为 str 的实例：

from transparentpath import Path
path = Path()
isinstance(path, str)  # returns True

这是允许的

from transparentpath import Path
path = Path()
with open(path, "w/r/a/b...") as ifile:
    ...

去工作。如果您想检查 path 是否实际上是一个 TransparentPath 而不是别的，请使用

from transparentpath import Path
path = Path()
assert type(path) == Path
assert issubclass(path, Path)

反而。

在 fsspec.implementations.local.LocalFileSystem、gcs.GCSFileSystem、pathlib.Path 或 str 中有效的任何方法或属性都可以在 TransparentPath 对象上使用。

警告

关于 GCS 行为的警告

如果您使用 GCS：

1. 请记住，目录不是 GCS 上的东西。

2. 您不需要 GCS 上文件的父目录来创建文件：如果它们不存在，则将创建它们（但这不是真正的本地化）。

3. 如果您删除一个单独存在于其父目录中的文件，这些目录就会消失。

4. 如果文件存在于与目录相同的路径，则 TransparentPath 无法知道哪个是文件，哪个是目录，并且会在对象创建时引发 TPMultipleExistenceError。如果外部源创建了文件/目录的副本，则几乎在每种方法中都会对多重性进行检查。这种情况不能在本地发生。但是，如果缓存不经常更新，它可能会在远程发生。执行此检查会显着增加计算时间（例如，如果在包含大量文件的目录上使用 glob）。您可以全局禁用它（TransparentPath._do_check = False 和 TransparentPath._do_update_cache = False），用于特定路径（在路径创建时传递 nockeck=True），或通过将 fast=True 作为附加参数传递给 glob 和 ls。

速度

GCS 上的 TransparentPath 很慢，因为验证多个存在和缓存更新。但是，可以稍微调整一下。如前所述，可以通过以下方式为所有路径停用缓存更新和多重存在检查

from transparentpath import TransparentPath
TransparentPath._do_update_cache = False
TransparentPath._do_check = False

它们也可以仅通过以下方式为一条路径停用

p = TransparentPath("somepath", nocheck=True, notupdatecache=True)

还可以指定何时进行这些检查：在创建路径时、路径使用情况（读取、写入、存在......）或两者兼而有之。可以在所有路径或仅部分路径上设置它：

TransparentPath._when_checked = {"created": True, "used": False}  # Default value
TransparentPath._when_updated = {"created": True, "used": False}  # Default value
p = TransparentPath(
  "somepath", when_checked={"created": False, "used": False}, notupdatecache={"created": False, "used": False}
)

检查和更新还有一个以秒为单位的过期时间：如果不是很久以前完成的操作，则该操作不会完成。这些过期时间默认为 1 秒，可以通过以下方式更改：

TransparentPath._check_expire = 10
TransparentPath._update_expire = 10
p = TransparentPath("somepath", check_expire=0, update_expire=0)

glob() 和 ls() 有自己的加速方式：

p.glob("/*", fast=True)
p.ls("", fast=True)

基本上，fast=True对于该方法找到的所有项目，意味着“不检查也不更新缓存”。

内置打开

Builtinopen()由 TransparentPath 重载以支持为其提供 TransparentPath。如果您未创建的包中的方法在 with 语句中使用open()，则一切都应该使用透明路径开箱即用。

但是，如果它使用 的输出，open()则必须创建一个类来覆盖此方法以及使用其输出的任何内容。确实，open()返回的是文件描述符，而不是 IO，而且我没有找到在 gcs 上访问文件描述符的方法。例如，在 FileLock 包中，acquire()方法调用调用的 _acquire()方法os.open()，所以我必须这样做：

from filelock import FileLock
from transparentpath import TransparentPath as Path

class MyFileLock(FileLock):
    def _acquire(self):
        tmp_lock_file = self._lock_file
        if not type(tmp_lock_file) == Path:
            tmp_lock_file = Path(tmp_lock_file)
        try:
            fd = tmp_lock_file.open("x")
        except (IOError, OSError, FileExistsError):
            pass
        else:
            self._lock_file_fd = fd
        return None

原来的方法是：

import os
...
def _acquire(self):
    open_mode = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
    try:
        fd = os.open(self._lock_file, open_mode)
    except (IOError, OSError):
        pass
    else:
        self._lock_file_fd = fd
    return None
...

我尝试实现在 pathlib.Path 或文件系统中有效的任何方法的工作版本，但是不会很快考虑到其中任何一个的未来变化。您可以通过打开问题来报告缺少的支持。