Enable PathNode and PickleNode to deal with URLs, S3, etc.. (#525)

Author: tobiasraabe

.pre-commit-config.yaml

@@ -99,6 +99,7 @@ repos:
                 docs/source/how_to_guides/using_task_returns.md|
                 docs/source/how_to_guides/writing_custom_nodes.md|
                 docs/source/how_to_guides/hashing_inputs_of_tasks.md|
+                docs/source/how_to_guides/remote_files.md|
                 docs/source/reference_guides/hookspecs.md|
                 docs/source/tutorials/configuration.md|
                 docs/source/tutorials/debugging.md|

docs/source/changes.md

@@ -15,7 +15,8 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and
   {func}`~pytask.task`. Closes {issue}`512`.
 - {pull}`522` improves the issue templates.
 - {pull}`523` refactors `_pytask.console._get_file`.
-- {pull}`524` improves some linting and formatter rules.
+- {pull}`524` improves some linting and formatting rules.
+- {pull}`525` enables pytask to work with remote files using universal_pathlib.
 
 ## 0.4.4 - 2023-12-04
 
@@ -25,14 +26,14 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and
 ## 0.4.3 - 2023-12-01
 
 - {pull}`483` simplifies the teardown of a task.
-- {pull}`484` raises more informative error when directories instead of files are used
-  with path nodes.
-- {pull}`485` adds missing steps to unconfigure pytask after the job is done which
+- {pull}`484` raises an informative error message when directories instead of files are
+  used with path nodes.
+- {pull}`485` adds missing steps to unconfigure pytask after the job is done, which
   caused flaky tests.
 - {pull}`486` adds default names to {class}`~pytask.PPathNode`.
 - {pull}`488` raises an error when an invalid value is used in a return annotation.
 - {pull}`489` and {pull}`491` simplifies parsing products and does not raise an error
-  when a product annotation is used with the argument name `produces`. And, allow
+  when a product annotation is used with the argument name `produces`. And allow
   `produces` to intake any node.
 - {pull}`490` refactors and better tests parsing of dependencies.
 - {pull}`493` allows tasks to depend on other tasks.
@@ -49,7 +50,7 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and
 
 ## 0.4.2 - 2023-11-08
 
-- {pull}`449` simplifies the code building the plugin manager.
+- {pull}`449` simplifies the code building of the plugin manager.
 - {pull}`451` improves `collect_command.py` and renames `graph.py` to `dag_command.py`.
 - {pull}`454` removes more `.svg`s and replaces them with animations.
 - {pull}`455` adds more explanation when {meth}`~pytask.PNode.load` fails during the
@@ -58,7 +59,7 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and
 - {pull}`457` refactors everything around formatting node names.
 - {pull}`459` adds a pre-commit hook to sort `__all__`.
 - {pull}`460` simplifies removing internal tracebacks from exceptions with a cause.
-- {pull}`463` raise error when a task function is not defined inside the loop body.
+- {pull}`463` raises an error when a task function is not defined inside the loop body.
 - {pull}`464` improves pinned dependencies.
 - {pull}`465` adds test to ensure internal tracebacks are removed by reports.
 - {pull}`466` implements hashing for files instead of modification timestamps.

docs/source/how_to_guides/index.md

@@ -13,6 +13,7 @@ maxdepth: 1
 ---
 migrating_from_scripts_to_pytask
 interfaces_for_dependencies_products
+remote_files
 functional_interface
 capture_warnings
 how_to_influence_build_order

docs/source/how_to_guides/remote_files.md

@@ -0,0 +1,92 @@
+# Remote files
+
+So far, we have only dealt with local files in the tutorials and guides. But there are
+lots of use cases to deal with remote files.
+
+- You distribute the workflow without the data and want to make it easy for others to
+  get started. So, some tasks reference remote files instead of local files.
+- You store the workflow results in remote storage to save and distribute them.
+
+pytask uses [universal_pathlib](https://github.com/fsspec/universal_pathlib) to work
+with remote files. The package provides a {mod}`pathlib`-like interface, making it very
+easy to interact with files from an HTTP(S)-, Dropbox-, S3-, GCP-, Azure-based
+filesystem, and many more.
+
+:::{warning}
+universal_pathlib does currently not support Python 3.12. To track progress, see [this
+PR](https://github.com/fsspec/universal_pathlib/pull/152) and check the [releases
+`>0.1.4`](https://github.com/fsspec/universal_pathlib/releases)
+:::
+
+## HTTP(S)-based filesystem
+
+As an example for dealing with an HTTP(S)-based filesystem, we will download the iris
+data set and save it as a CSV file.
+
+```{literalinclude} ../../../docs_src/how_to_guides/remote_files/https.py
+```
+
+## Other filesystems
+
+universal_pathlib supports Azure Storage, Dropbox, Google Cloud Storage, AWS S3, and
+[many more filesystems](https://github.com/fsspec/universal_pathlib#currently-supported-filesystems-and-schemes).
+
+For example, let us try accessing a file in an S3 bucket. We pass `anon=True` to
+{class}`~upath.UPath` since no credentials are required.
+
+```pycon
+>>> from upath import UPath
+>>> path = UPath("s3://upath-aws-example/iris.data", anon=True)
+>>> path.stat()
+ModuleNotFoundError
+...
+ImportError: Install s3fs to access S3
+```
+
+Some filesystems are supported
+[out-of-the-box](https://filesystem-spec.readthedocs.io/en/latest/api.html#built-in-implementations).
+[Others](https://filesystem-spec.readthedocs.io/en/latest/api.html#other-known-implementations)
+are available as plugins and require additional packages.
+
+After installing s3fs, rerun the command.
+
+```pycon
+>>> path.stat()
+{'ETag': '"42615765a885ddf54427f12c34a0a070"',
+ 'LastModified': datetime.datetime(2023, 12, 11, 23, 50, 3, tzinfo=tzutc()),
+ 'size': 4551,
+ 'name': 'upath-aws-example/iris.data',
+ 'type': 'file',
+ 'StorageClass': 'STANDARD',
+ 'VersionId': None,
+ 'ContentType': 'binary/octet-stream'}
+```
+
+Usually, you will need credentials to access files. Search in
+[fsspec's documentation](https://filesystem-spec.readthedocs.io/en/latest)
+or the plugin's documentation, here
+[s3fs](https://s3fs.readthedocs.io/en/latest/#credentials), for information on
+authentication. One way would be to set the environment variables `AWS_ACCESS_KEY_ID`
+and `AWS_SECRET_ACCESS_KEY`.
+
+## Detecting changes in remote files
+
+pytask uses the [entity tag (ETag)](https://en.wikipedia.org/wiki/HTTP_ETag) to detect
+changes in remote files. The ETag is an optional header field that can signal a file has
+changed. For example,
+[AWS S3 uses an MD5 digest](https://teppen.io/2018/06/23/aws_s3_etags/) of the uploaded
+file as the ETag. If the file changes, so does the ETag, and pytask will detect it.
+
+Many files on the web do not provide an ETag like this version of the iris dataset.
+
+```pycon
+>>> import requests
+>>> url = "https://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data"
+>>> r = requests.head(url)
+>>> r.headers
+{'Server': 'nginx/1.25.3', 'Date': 'Sun, 10 Dec 2023 23:59:21 GMT', 'Connection': 'keep-alive'}
+```
+
+In these instances, pytask does not recognize if the file has changed and only reruns
+the task if other conditions are not met, like the product is missing, the task module
+has changed, etc..

docs/source/tutorials/defining_dependencies_products.md

@@ -14,6 +14,10 @@ If you want to avoid type annotations for now, look at the tab named `produces`.
 The `Decorators` tab documents the deprecated approach that should not be used
 anymore and will be removed in version v0.5.
 
+In this tutorial, we only deal with local files. If you want to use pytask with files
+online, S3, GCP, Azure, etc., read the
+{doc}`guide on remote files <../how_to_guides/remote_files>`.
+
 First, we focus on defining products that should already be familiar to you. Then,
 we focus on how you can declare task dependencies.
 

docs_src/how_to_guides/remote_files/https.py

@@ -0,0 +1,11 @@
+from pathlib import Path
+from typing import Annotated
+
+from upath import UPath
+
+
+url = UPath("https://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data")
+
+
+def task_download_file(path: UPath = url) -> Annotated[str, Path("data.csv")]:
+    return path.read_text()

environment.yml

@@ -21,6 +21,7 @@ dependencies:
   - sqlalchemy >=1.4.36
   - tomli >=1.0.0
   - typing_extensions
+  - universal_pathlib
 
   # Misc
   - deepdiff

pyproject.toml

@@ -57,6 +57,7 @@ name = "Tobias Raabe"
 email = "[email protected]"
 
 [project.optional-dependencies]
+all = ['universal-pathlib; python_version<"3.12"']
 docs = [
   "furo",
   "ipython",
@@ -76,6 +77,9 @@ test = [
   "pytest-cov",
   "pytest-xdist",
   "syrupy",
+  # For HTTPPath tests.
+  "aiohttp",
+  "requests",
 ]
 
 [project.urls]

src/_pytask/collect.py

@@ -41,6 +41,14 @@
 from _pytask.typing import is_task_function
 from rich.text import Text
 
+try:
+    from upath import UPath
+except ImportError:  # pragma: no cover
+
+    class UPath:  # type: ignore[no-redef]
+        ...
+
+
 if TYPE_CHECKING:
     from _pytask.session import Session
     from _pytask.models import NodeInfo
@@ -310,7 +318,7 @@ def pytask_collect_task(
 
 
 @hookimpl(trylast=True)
-def pytask_collect_node(session: Session, path: Path, node_info: NodeInfo) -> PNode:  # noqa: C901
+def pytask_collect_node(session: Session, path: Path, node_info: NodeInfo) -> PNode:  # noqa: C901, PLR0912
     """Collect a node of a task as a :class:`pytask.PNode`.
 
     Strings are assumed to be paths. This might be a strict assumption, but since this
@@ -354,12 +362,24 @@ def pytask_collect_node(session: Session, path: Path, node_info: NodeInfo) -> PN
             node.path, session.config["paths"] or (session.config["root"],)
         )
 
-    if isinstance(node, PPathNode) and node.path.is_dir():
+    # Skip ``is_dir`` for remote UPaths because it downloads the file and blocks the
+    # collection.
+    if (
+        isinstance(node, PPathNode)
+        and not isinstance(node.path, UPath)
+        and node.path.is_dir()
+    ):
         raise ValueError(_TEMPLATE_ERROR_DIRECTORY.format(path=node.path))
 
     if isinstance(node, PNode):
         return node
 
+    if isinstance(node, UPath):
+        if not node.protocol:
+            node = Path(node)
+        else:
+            return PathNode(name=node.name, path=node)
+
     if isinstance(node, Path):
         if not node.is_absolute():
             node = path.joinpath(node)

src/_pytask/nodes.py

@@ -4,6 +4,7 @@
 import hashlib
 import inspect
 import pickle
+from os import stat_result
 from pathlib import Path  # noqa: TCH003
 from typing import Any
 from typing import Callable
@@ -178,8 +179,14 @@ def state(self) -> str | None:
 
         """
         if self.path.exists():
-            modification_time = self.path.stat().st_mtime
-            return hash_path(self.path, modification_time)
+            stat = self.path.stat()
+            if isinstance(stat, stat_result):
+                modification_time = self.path.stat().st_mtime
+                return hash_path(self.path, modification_time)
+            if isinstance(stat, dict):
+                return stat.get("ETag", "0")
+            msg = "Unknown stat object."
+            raise NotImplementedError(msg)
         return None
 
     def load(self, is_product: bool = False) -> Path:  # noqa: ARG002
@@ -317,8 +324,14 @@ def from_path(cls, path: Path) -> PickleNode:
 
     def state(self) -> str | None:
         if self.path.exists():
-            modification_time = self.path.stat().st_mtime
-            return hash_path(self.path, modification_time)
+            stat = self.path.stat()
+            if isinstance(stat, stat_result):
+                modification_time = self.path.stat().st_mtime
+                return hash_path(self.path, modification_time)
+            if isinstance(stat, dict):
+                return stat.get("ETag", "0")
+            msg = "Unknown stat object."
+            raise NotImplementedError(msg)
         return None
 
     def load(self, is_product: bool = False) -> Any:

src/_pytask/path.py

@@ -50,7 +50,9 @@ def relative_to(path: Path, source: Path, include_source: bool = True) -> Path:
     return Path(source_name, path.relative_to(source))
 
 
-def find_closest_ancestor(path: Path, potential_ancestors: Sequence[Path]) -> Path:
+def find_closest_ancestor(
+    path: Path, potential_ancestors: Sequence[Path]
+) -> Path | None:
     """Find the closest ancestor of a path.
 
     In case a path is the path to the task file itself, we return the path.
@@ -76,10 +78,19 @@ def find_closest_ancestor(path: Path, potential_ancestors: Sequence[Path]) -> Pa
         if ancestor == path:
             return path
 
-        candidate = find_common_ancestor(path, ancestor)
-        potential_closest_ancestors.append(candidate)
-
-    return sorted(potential_closest_ancestors, key=lambda x: len(x.parts))[-1]
+        with contextlib.suppress(ValueError):
+            candidate = find_common_ancestor(path, ancestor)
+            potential_closest_ancestors.append(candidate)
+
+    return next(
+        (
+            i
+            for i in sorted(
+                potential_closest_ancestors, key=lambda x: len(x.parts), reverse=True
+            )
+        ),
+        None,
+    )
 
 
 def find_common_ancestor(*paths: Path) -> Path:

tests/test_execute.py

@@ -1027,3 +1027,22 @@ def func(path): path.touch()
     result = runner.invoke(cli, [tmp_path.as_posix()])
     assert result.exit_code == ExitCode.OK
     assert tmp_path.joinpath("out.txt").exists()
+
+
+@pytest.mark.skipif(sys.version_info >= (3, 12), reason="Not supported in Python 3.12.")
+def test_with_http_path(runner, tmp_path):
+    source = """
+    from upath import UPath
+    from typing_extensions import Annotated
+
+    url = "https://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data"
+
+    def task_example(path = UPath(url)) -> Annotated[str, UPath("data.txt")]:
+        return path.read_text()
+    """
+    tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(source))
+
+    result = runner.invoke(cli, [tmp_path.as_posix()])
+    print(result.output)  # noqa: T201
+    assert result.exit_code == ExitCode.OK
+    assert tmp_path.joinpath("data.txt").exists()

tox.ini

@@ -6,9 +6,9 @@ passenv = CI
 usedevelop = true
 
 [testenv:pytest]
-extras = test
+extras = all, test
 deps =
     pygraphviz;platform_system != "Windows"
 
 commands =
-    pytest {posargs} -vv
+    pytest {posargs}