From 9b5ab838031459b6892db0207c8200f939668d50 Mon Sep 17 00:00:00 2001 From: Bolke de Bruin Date: Mon, 13 Nov 2023 19:15:05 +0100 Subject: [PATCH 01/12] Refactor ObjectSToragePath to be a Path --- airflow/io/implementations.py | 28 ++ airflow/io/registry.py | 103 ++++++ airflow/io/store/path.py | 594 ++++++++-------------------------- tests/io/store/test_store.py | 81 ++--- 4 files changed, 281 insertions(+), 525 deletions(-) create mode 100644 airflow/io/implementations.py create mode 100644 airflow/io/registry.py diff --git a/airflow/io/implementations.py b/airflow/io/implementations.py new file mode 100644 index 0000000000000..db492ad52b4a4 --- /dev/null +++ b/airflow/io/implementations.py @@ -0,0 +1,28 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +from upath.implementations.cloud import CloudPath + +from airflow.io.store.path import ObjectStoragePath, _AirflowCloudAccessor + + +class AirflowCloudPath(ObjectStoragePath, CloudPath): + """A Path implementation for that uses Airflow's Connections.""" + + _default_accessor = _AirflowCloudAccessor diff --git a/airflow/io/registry.py b/airflow/io/registry.py new file mode 100644 index 0000000000000..7cbf94d6152e8 --- /dev/null +++ b/airflow/io/registry.py @@ -0,0 +1,103 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import sys +from collections import ChainMap +from functools import lru_cache +from importlib import import_module +from importlib.metadata import entry_points +from typing import TYPE_CHECKING, MutableMapping + +import upath + +if TYPE_CHECKING: + import airflow.io.store.path + +_ENTRY_POINT_GROUP = "airflow_pathlib.implementations" + + +class _Registry(MutableMapping[str, "type[airflow.io.store.path.ObjectStoragePath]"]): + """Registry for ObjectStoragePath subclasses.""" + + # _default_ set for serialization + known_implementations: dict[str, str] = {"_default_": "airflow.io.implementations.AirflowCloudPath"} + + def __init__(self) -> None: + if sys.version_info >= (3, 10): + eps = entry_points(group=_ENTRY_POINT_GROUP) + else: + eps = entry_points().get(_ENTRY_POINT_GROUP, []) + self._entries = {ep.name: ep for ep in eps} + self._m = ChainMap({}, self.known_implementations) # type: ignore + + def __getitem__(self, item: str) -> type[airflow.io.store.path.ObjectStoragePath]: + fqn = self._m.get(item) + if fqn is None: + if item in self._entries: + fqn = self._m[item] = self._entries[item].load() + if fqn is None: + raise KeyError(f"{item} not in registry") + if isinstance(fqn, str): + module_name, name = fqn.rsplit(".", 1) + mod = import_module(module_name) + cls = getattr(mod, name) # type: ignore + else: + cls = fqn + return cls + + def __setitem__(self, item: str, value: type[upath.core.UPath] | str) -> None: + if not ((isinstance(value, type) and issubclass(value, upath.core.UPath)) or isinstance(value, str)): + raise ValueError(f"expected ObjectStorage subclass or FQN-string, got: {type(value).__name__!r}") + self._m[item] = value + + def __delitem__(self, __v: str) -> None: + raise NotImplementedError("removal is unsupported") + + def __len__(self) -> int: + return len(set().union(self._m, self._entries)) + + def __iter__(self): + return iter(set().union(self._m, self._entries)) + + +registry = _Registry() + + +@lru_cache +def get_path_class( + protocol: str, *, default: str | None = None +) -> type[upath.core.UPath | airflow.io.store.path.ObjectStoragePath]: + """Return the path cls for the given protocol. + + Returns `None` or ``default`` if no matching protocol can be found. + + Parameters + ---------- + protocol: + The protocol string + default: + returned if no matching protocol can be found + """ + try: + return registry[protocol] + except KeyError: + if default: + return registry[default] + else: + raise diff --git a/airflow/io/store/path.py b/airflow/io/store/path.py index 17e951886f838..77e74d3b7422c 100644 --- a/airflow/io/store/path.py +++ b/airflow/io/store/path.py @@ -21,41 +21,66 @@ import os import shutil import typing -from io import UnsupportedOperation -from stat import S_ISLNK +from pathlib import PurePath +from urllib.parse import SplitResult, urlsplit +from fsspec.core import split_protocol from fsspec.utils import stringify_path +from upath import UPath +from upath.implementations.cloud import _CloudAccessor +from upath.registry import get_upath_class -from airflow.io.store import ObjectStore, attach +from airflow.io.registry import get_path_class +from airflow.io.store import attach from airflow.io.store.stat import stat_result if typing.TYPE_CHECKING: from fsspec import AbstractFileSystem -def _rewrite_info(info: dict, store: ObjectStore) -> dict: - info["name"] = ObjectStoragePath(info["name"], store=store) - return info +PT = typing.TypeVar("PT", bound="ObjectStoragePath") +default = "file" -@functools.total_ordering -class ObjectStoragePath(os.PathLike): + +class _AirflowCloudAccessor(_CloudAccessor): + __slots__ = ("_store", "_conn_id") + + def __init__(self, parsed_url: SplitResult | None, **kwargs: typing.Any) -> None: + _store = kwargs.pop("store", None) + conn_id = kwargs.pop("conn_id", None) + if _store: + self._store = _store + elif parsed_url and parsed_url.scheme: + self._store = attach(parsed_url.scheme, conn_id) # todo add kwargs as storage_options + else: + self._store = attach(default, conn_id) + self._conn_id = conn_id + + @property + def _fs(self) -> AbstractFileSystem: + return self._store.fs + + def __eq__(self, other): + return isinstance(other, _AirflowCloudAccessor) and self._store == other._store + + +class ObjectStoragePath(UPath): """A path-like object for object storage.""" __version__: typing.ClassVar[int] = 1 + _default_accessor = _AirflowCloudAccessor + sep: typing.ClassVar[str] = "/" root_marker: typing.ClassVar[str] = "/" - _store: ObjectStore | None _bucket: str _key: str - _conn_id: str | None _protocol: str _hash: int | None __slots__ = ( - "_store", "_bucket", "_key", "_conn_id", @@ -63,171 +88,85 @@ class ObjectStoragePath(os.PathLike): "_hash", ) - def __init__( - self, - path: str | ObjectStoragePath, - *, - conn_id: str | None = None, - store: ObjectStore | None = None, - ) -> None: - self._conn_id = conn_id - self._store = store - - if isinstance(path, ObjectStoragePath): - self._protocol = path._protocol - self._bucket = path._bucket - self._key = path._key - self._store = path._store + def __new__(cls: type[PT], *args: str | os.PathLike, **kwargs: typing.Any) -> PT: + args_list = list(args) + try: + other = args_list.pop(0) + except IndexError: + other = "." else: - self._protocol, self._bucket, self._key = self._split_path(path) - - if store: - self._conn_id = store.conn_id - self._protocol = self._protocol if self._protocol else store.protocol - elif self._protocol and not self._store: - self._store = attach(self._protocol, conn_id) - - @classmethod - def _split_path(cls, p: typing.Any) -> tuple[str, str, str]: - protocol, _, path = str(stringify_path(p)).rpartition("://") - - if cls.sep not in path: - bucket = path - key = "" + other = other or "." + + if isinstance(other, PurePath): + _cls: type[typing.Any] = type(other) + drv, root, parts = _cls._parse_args(args_list) + drv, root, parts = _cls._flavour.join_parsed_parts( + other._drv, # type: ignore[attr-defined] + other._root, # type: ignore[attr-defined] + other._parts, # type: ignore[attr-defined] + drv, + root, + parts, # type: ignore + ) + + _kwargs = getattr(other, "_kwargs", {}) + _url = getattr(other, "_url", None) + other_kwargs = _kwargs.copy() + if _url and _url.scheme: + other_kwargs["url"] = _url + new_kwargs = _kwargs.copy() + new_kwargs.update(kwargs) + + return _cls(_cls._format_parsed_parts(drv, root, parts, **other_kwargs), **new_kwargs) + + url = stringify_path(other) + protocol, _ = split_protocol(url) + parsed_url = urlsplit(url) + + if protocol is None: + protocol = kwargs.get("scheme", parsed_url.scheme) else: - bucket, key = path.split(cls.sep, 1) + protocol = kwargs.get("scheme", protocol) - # we don't care about versions etc - return protocol, bucket, key + path_cls = get_path_class(protocol, default="_default_") - def __repr__(self) -> str: - return f"<{type(self).__name__}('{self}')>" + for key in ["scheme", "netloc"]: + val = kwargs.get(key) + if val: + parsed_url = parsed_url._replace(**{key: val}) - def __str__(self) -> str: - return ( - f"{self._protocol}://{self._bucket}/{self._key}" - if self._protocol - else f"{self._bucket}/{self._key}" - ) - - __fspath__ = __str__ + if not parsed_url.path: + parsed_url = parsed_url._replace(path="/") # ensure path has root - def __lt__(self, other: typing.Any) -> bool: - if not isinstance(other, ObjectStoragePath): - return NotImplemented - return self._bucket < other._bucket - - def __eq__(self, other: typing.Any) -> bool: - if not isinstance(other, ObjectStoragePath): - return NotImplemented - return self._bucket == other._bucket + if not protocol: + args_list.insert(0, url) + else: + args_list.insert(0, parsed_url.path) - def __ne__(self, other: typing.Any) -> bool: - if not isinstance(other, ObjectStoragePath): - return NotImplemented - return self._bucket != other._bucket + return path_cls._from_parts(args_list, url=parsed_url, **kwargs) # type: ignore @functools.lru_cache def __hash__(self) -> int: return hash(self._bucket) - def __truediv__(self, other: typing.Any) -> ObjectStoragePath: - o_protocol, o_bucket, _ = self._split_path(other) - if isinstance(other, ObjectStoragePath) and o_bucket and self._bucket != o_bucket: - raise ValueError("Cannot combine paths from different buckets / containers") - - if o_protocol and self._protocol != o_protocol: - raise ValueError("Cannot combine paths from different protocols") - - self_path = str(stringify_path(self)) - other_path = str(stringify_path(other)) - - path = f"{self_path.rstrip(self.sep)}/{other_path.lstrip(self.sep)}" - return ObjectStoragePath(path, conn_id=self._conn_id) - - def _unsupported(self, method_name: str) -> typing.NoReturn: - msg = f"{type(self).__name__}.{method_name}() is unsupported" - raise UnsupportedOperation(msg) - def samestore(self, other: typing.Any) -> bool: - return isinstance(other, ObjectStoragePath) and self._store == other._store + return isinstance(other, ObjectStoragePath) and self._accessor == other._accessor @property def container(self) -> str: - return self._bucket + return self.bucket @property def bucket(self) -> str: - return self._bucket + return self._url.netloc @property def key(self) -> str: - return self._key - - @property - def store(self) -> ObjectStore: - if not self._store: - raise ValueError("Cannot do operations. No store attached.") - - return self._store - - @property - def protocol(self) -> str: - return self._protocol - - @property - def fs(self) -> AbstractFileSystem: - return self.store.fs - - @property - def parent(self) -> ObjectStoragePath: - return ObjectStoragePath(self.store.fs._parent(str(self)), store=self.store) + return self._url.path def stat(self, *, follow_symlinks: bool = True) -> stat_result: """Return the result of the `stat()` call.""" # noqa: D402 - return stat_result(self.store.fs.stat(self), protocol=self.store.protocol, conn_id=self.store.conn_id) - - def lstat(self) -> stat_result: - """Like stat() except that it doesn't follow symlinks.""" - return self.stat(follow_symlinks=False) - - def exists(self) -> bool: - """Whether this path exists.""" - return self.store.fs.exists(self) - - def is_dir(self) -> bool: - """Return True if this path is directory like.""" - return self.store.fs.isdir(self) - - def is_file(self) -> bool: - """Return True if this path is a regular file.""" - return self.store.fs.isfile(self) - - def is_mount(self) -> bool: - self._unsupported("is_mount") - - def is_symlink(self) -> bool: - """Whether this path is a symbolic link.""" - try: - return S_ISLNK(self.lstat().st_mode) - except OSError: - # Path doesn't exist - return False - except ValueError: - # Non-encodable path - return False - - def is_block_device(self) -> bool: - self._unsupported("is_block_device") - - def is_char_device(self) -> bool: - self._unsupported("is_char_device") - - def is_fifo(self) -> bool: - self._unsupported("is_fifo") - - def is_socket(self) -> bool: - self._unsupported("is_socket") + return stat_result(self._accessor.stat(self), protocol=self.protocol, conn_id=self._accessor.conn_id) def samefile(self, other_path: typing.Any) -> bool: """Return whether other_path is the same or not as this file.""" @@ -243,250 +182,11 @@ def samefile(self, other_path: typing.Any) -> bool: and st["ino"] == other_st["ino"] ) - def checksum(self) -> int: - """Return the checksum of the file at this path.""" - return self.store.fs.checksum(self) - - def open( - self, - mode="rb", - block_size=None, - cache_options=None, - compression=None, - encoding=None, - errors=None, - newline=None, - **kwargs, - ) -> typing.IO: - """ - Return a file-like object from the filesystem. - - The resultant instance must function correctly in a context 'with' block. - - :param mode: str like 'rb', 'w' - See builtin 'open()'. - :param block_size: int - Some indication of buffering - this is a value in bytes. - :param cache_options: dict, optional - Extra arguments to pass through to the cache. - :param compression: string or None - If given, open file using a compression codec. Can either be a compression - name (a key in 'fsspec.compression.compr') or 'infer' to guess the - compression from the filename suffix. - :param encoding: passed on to TextIOWrapper for text mode - :param errors: passed on to TextIOWrapper for text mode - :param newline: passed on to TextIOWrapper for text mode - - kwargs: Additional keyword arguments to be passed on. - """ - return self.store.fs.open( - str(self), - mode=mode, - block_size=block_size, - cache_options=cache_options, - compression=compression, - encoding=encoding, - errors=errors, - newline=newline, - **kwargs, - ) - - def read_bytes(self, start: int | None = None, end: int | None = None) -> bytes: - """Open the file in bytes mode, read it, and close the file.""" - return self.store.fs.read_bytes(str(self), start=start, end=end) - - def read_text(self, encoding=None, errors=None, newline=None, **kwargs) -> str: - """Open the file in text mode, read it, and close the file.""" - return self.store.fs.read_text(str(self), encoding=encoding, errors=errors, newline=newline, **kwargs) - - def write_bytes(self, data, **kwargs) -> int: - """Open the file in bytes mode, write to it, and close the file.""" - return self.store.fs.pipe_file(self, value=data, **kwargs) - - def write_text(self, data, encoding=None, errors=None, newline=None, **kwargs) -> int: - """Open the file in text mode, write to it, and close the file.""" - return self.store.fs.write_text( - str(self), value=data, encoding=encoding, errors=errors, newline=newline, **kwargs - ) - - def iterdir(self): - """Iterate over the files in this directory.""" - self._unsupported("iterdir") - def _scandir(self): # Emulate os.scandir(), which returns an object that can be used as a # context manager. return contextlib.nullcontext(self.iterdir()) - def glob(self, pattern: str, maxdepth: int | None = None, **kwargs): - """ - Find files by glob-matching. - - If the path ends with '/', only folders are returned. - - We support ``"**"``, - ``"?"`` and ``"[..]"``. We do not support ^ for pattern negation. - - The `maxdepth` option is applied on the first `**` found in the path. - - Search path names that contain embedded characters special to this - implementation of glob may not produce expected results; - e.g., 'foo/bar/*starredfilename*'. - - :param pattern: str - The glob pattern to match against. - :param maxdepth: int or None - The maximum depth to search. If None, there is no depth limit. - - kwargs: Additional keyword arguments to be passed on. - """ - path = os.path.join(self._bucket, pattern) - - detail = kwargs.get("detail", False) - items = self.store.fs.glob(path, maxdepth=maxdepth, **kwargs) - if detail: - t = { - ObjectStoragePath(k, store=self.store): _rewrite_info(v, self.store) for k, v in items.items() - } - return t - else: - return [ObjectStoragePath(c, store=self.store) for c in items] - - def rglob(self, maxdepth: int | None = None, **kwargs): - self._unsupported("rglob") - - def walk(self, maxdepth: int | None = None, topdown: bool = True, on_error: str = "omit", **kwargs): - """ - Return all files belows path. - - List all files, recursing into subdirectories; output is iterator-style, - like ``os.walk()``. For a simple list of files, ``find()`` is available. - - When topdown is True, the caller can modify the dirnames list in-place (perhaps - using del or slice assignment), and walk() will - only recurse into the subdirectories whose names remain in dirnames; - this can be used to prune the search, impose a specific order of visiting, - or even to inform walk() about directories the caller creates or renames before - it resumes walk() again. - Modifying dirnames when topdown is False has no effect. (see os.walk) - - Note that the "files" outputted will include anything that is not - a directory, such as links. - - :param maxdepth: int or None - Maximum recursion depth. None means limitless, but not recommended - on link-based file-systems. - :param topdown: bool (True) - Whether to walk the directory tree from the top downwards or from - the bottom upwards. - :param on_error: "omit", "raise", a collable - if omit (default), path with exception will simply be empty; - If raise, an underlying exception will be raised; - if callable, it will be called with a single OSError instance as argument - kwargs: Additional keyword arguments to be passed on. - """ - detail = kwargs.get("detail", False) - items = self.store.fs.walk(str(self), maxdepth=maxdepth, topdown=topdown, on_error=on_error, **kwargs) - if not detail: - for path, dirs, files in items: - yield ObjectStoragePath(path, store=self.store), dirs, files - else: - for path, dirs, files in items: - yield ( - ObjectStoragePath(path, store=self.store), - {k: _rewrite_info(v, self.store) for k, v in dirs.items()}, - {k: _rewrite_info(v, self.store) for k, v in files.items()}, - ) - - def ls(self, detail: bool = True, **kwargs) -> list[ObjectStoragePath] | list[dict]: - """ - List files at path. - - :param detail: bool - If True, return a dict containing details about each entry, otherwise - return a list of paths. - - kwargs: Additional keyword arguments to be passed on. - """ - items = self.store.fs.ls(str(self), detail=detail, **kwargs) - - if detail: - return [_rewrite_info(c, self.store) for c in items] - else: - return [ObjectStoragePath(c, store=self.store) for c in items] - - def absolute(self) -> ObjectStoragePath: - """Return an absolute version of this path. Resolving any aliases.""" - return ObjectStoragePath(f"{self.store.protocol}://{self._key}") - - def touch(self, truncate: bool = True) -> None: - """Create an empty file, or update the timestamp. - - :param truncate: bool (True) - If True, always set the file size to 0; if False, update the timestamp and - leave the file unchanged, if the backend allows this. - """ - self.store.fs.touch(str(self), truncate=truncate) - - def mkdir(self, create_parents: bool = True, exists_ok: bool = False, **kwargs) -> None: - """ - Create a directory entry at the specified path or within a bucket/container. - - For systems that don't have true directories, it may create a directory entry - for this instance only and not affect the real filesystem. - - :param create_parents: bool - if True, this is equivalent to 'makedirs'. - :param exists_ok: bool - if True, do not raise an error if the target directory already exists. - - kwargs: Additional keyword arguments, which may include permissions, etc. - """ - if not exists_ok and self.exists(): - raise FileExistsError(f"Target {self} exists") - with contextlib.suppress(FileExistsError): - self.store.fs.mkdir(str(self), create_parents=create_parents, **kwargs) - - def unlink(self, recursive: bool = False, maxdepth: int | None = None) -> None: - """ - Remove this file or link. - - If the path is a directory, use rmdir() instead. - """ - self.store.fs.rm(str(self), recursive=recursive, maxdepth=maxdepth) - - rm = unlink - """ - Remove this file or link. - - Alias of unlink - """ - - def rmdir(self) -> None: - """Remove this directory. The directory must be empty.""" - self.store.fs.rmdir(str(self)) - - def rename(self, target: str | ObjectStoragePath, overwrite=False) -> ObjectStoragePath: - """ - Rename this path to the target path. - - The target path may be absolute or relative. Relative paths are - interpreted relative to the current working directory, *not* the - directory of the Path object. - - Returns the new Path instance pointing to the target path. - """ - if isinstance(target, str): - target = ObjectStoragePath(target, store=self.store) - - if not self.samestore(target): - raise ValueError("You can only rename within the same store") - - if not overwrite and self.store.fs.exists(target): - raise FileExistsError(f"Target {target} exists") - - return ObjectStoragePath(self.store.fs.mv(str(self), target), store=self._store) - def replace(self, target: str | ObjectStoragePath) -> ObjectStoragePath: """ Rename this path to the target path, overwriting if that path exists. @@ -499,11 +199,30 @@ def replace(self, target: str | ObjectStoragePath) -> ObjectStoragePath: """ return self.rename(target, overwrite=True) + @classmethod + def cwd(cls): + if cls is ObjectStoragePath: + return get_upath_class("").cwd() + else: + raise NotImplementedError + + @classmethod + def home(cls): + if cls is ObjectStoragePath: + return get_upath_class("").home() + else: + raise NotImplementedError + # EXTENDED OPERATIONS def ukey(self) -> str: """Hash of file properties, to tell if it has changed.""" - return self.store.fs.ukey(str(self)) + return self.fs.ukey(self.path) + + def checksum(self) -> int: + """Return the checksum of the file at this path.""" + # we directly access the fs here to avoid changing the abstract interface + return self.fs.checksum(self.path) def read_block(self, offset: int, length: int, delimiter=None): r"""Read a block of bytes. @@ -538,7 +257,7 @@ def read_block(self, offset: int, length: int, delimiter=None): -------- :func:`fsspec.utils.read_block` """ - return self.store.fs.read_block(str(self), offset, length, delimiter=delimiter) + return self.fs.read_block(self.path, offset, length, delimiter=delimiter) def sign(self, expiration: int = 100, **kwargs): """Create a signed URL representing the given path. @@ -556,68 +275,17 @@ def sign(self, expiration: int = 100, **kwargs): :raises NotImplementedError: if the method is not implemented for a store """ - return self.store.fs.sign(str(self), expiration=expiration, **kwargs) + return self.fs.sign(self.path, expiration=expiration, **kwargs) def size(self) -> int: """Size in bytes of the file at this path.""" - return self.store.fs.size(self) - - def du(self, total: bool = True, maxdepth: int | None = None, withdirs: bool = False, **kwargs): - """Space used by files and optionally directories within a path. - - Directory size does not include the size of its contents. - - :param total: bool - Whether to sum all the file sizes - :param maxdepth: int or None - Maximum number of directory levels to descend, None for unlimited. - :param withdirs: bool - Whether to include directory paths in the output. - - kwargs: Additional keyword arguments to be passed on. - - :returns: Dict of {path: size} if total=False, or int otherwise, where numbers - refer to bytes used. - """ - return self.store.fs.du(str(self), total=total, maxdepth=maxdepth, withdirs=withdirs, **kwargs) - - def find( - self, path: str, maxdepth: int | None = None, withdirs: bool = False, detail: bool = False, **kwargs - ): - """List all files below the specified path. - - Like posix ``find`` command without conditions. + return self.fs.size(self.path) - :param path: str - Path pattern to search. - :param maxdepth: int or None - If not None, the maximum number of levels to descend. - :param withdirs: bool - Whether to include directory paths in the output. This is True - when used by glob, but users usually only want files. - :param detail: bool - Whether to include file info. - - kwargs: Additional keyword arguments to be passed to ``ls``. - """ - path = self.sep.join([str(self), path.lstrip("/")]) - items = self.store.fs.find(path, maxdepth=maxdepth, withdirs=withdirs, detail=detail, **kwargs) - - if detail: - return { - ObjectStoragePath(k, store=self.store): _rewrite_info(v, self.store) for k, v in items.items() - } - else: - return [ObjectStoragePath(c, store=self.store) for c in items] - - def _cp_file(self, dst: str | ObjectStoragePath, **kwargs): + def _cp_file(self, dst: ObjectStoragePath, **kwargs): """Copy a single file from this path to another location by streaming the data.""" - if isinstance(dst, str): - dst = ObjectStoragePath(dst) - # create the directory or bucket if required if dst.key.endswith(self.sep) or not dst.key: - dst.mkdir(exists_ok=True, create_parents=True) + dst.mkdir(exist_ok=True, parents=True) dst = dst / self.key elif dst.is_dir(): dst = dst / self.key @@ -644,18 +312,16 @@ def copy(self, dst: str | ObjectStoragePath, recursive: bool = False, **kwargs) # same -> same if self.samestore(dst): - self.store.fs.copy(str(self), dst, recursive=recursive, **kwargs) + self.fs.copy(self.path, dst.path, recursive=recursive, **kwargs) return # use optimized path for local -> remote or remote -> local - if self.store.protocol == "file": - lpath = self.store.fs._strip_protocol(str(self)) - dst.store.fs.put(lpath, str(dst), recursive=recursive, **kwargs) + if self.protocol == "file": + dst.fs.put(self.path, dst.path, recursive=recursive, **kwargs) return - if dst.store.protocol == "file": - rpath = dst.store.fs._strip_protocol(str(dst)) - self.store.fs.get(str(self), rpath, recursive=recursive, **kwargs) + if dst.protocol == "file": + self.fs.get(self.path, dst.path, recursive=recursive, **kwargs) return if not self.exists(): @@ -666,18 +332,17 @@ def copy(self, dst: str | ObjectStoragePath, recursive: bool = False, **kwargs) if dst.is_file(): raise ValueError("Cannot copy directory to a file.") - dst.mkdir(exists_ok=True, create_parents=True) + dst.mkdir(exist_ok=True, parents=True) - out = self.store.fs.expand_path(str(self), recursive=True, **kwargs) - source_stripped = self.store.fs._strip_protocol(str(self)) + out = self.fs.expand_path(self.path, recursive=True, **kwargs) for path in out: # this check prevents one extra call to is_dir() as # glob returns self as well - if path == source_stripped: + if path == self.path: continue - src_obj = ObjectStoragePath(path, store=self.store) + src_obj = ObjectStoragePath(path, store=self._accessor._store) # skip directories, empty directories will not be created if src_obj.is_dir(): @@ -705,16 +370,16 @@ def move(self, path: str | ObjectStoragePath, recursive: bool = False, **kwargs) path = ObjectStoragePath(path) if self.samestore(path): - return self.store.fs.move(str(self), str(path), recursive=recursive, **kwargs) + return self.fs.move(self.path, path.path, recursive=recursive, **kwargs) # non-local copy self.copy(path, recursive=recursive, **kwargs) - self.unlink(recursive=recursive) + self.unlink() - def serialize(self) -> dict[str, str | ObjectStore]: + def serialize(self) -> dict[str, str]: return { "path": str(self), - "store": self.store, + **self._kwargs, } @classmethod @@ -722,4 +387,5 @@ def deserialize(cls, data: dict, version: int) -> ObjectStoragePath: if version > cls.__version__: raise ValueError(f"Cannot deserialize version {version} with version {cls.__version__}.") - return ObjectStoragePath(**data) + path = data.pop("path") + return ObjectStoragePath(path, **data) diff --git a/tests/io/store/test_store.py b/tests/io/store/test_store.py index 7a7018e7ecd34..6b738424d2a9c 100644 --- a/tests/io/store/test_store.py +++ b/tests/io/store/test_store.py @@ -58,19 +58,18 @@ def test_alias(self): def test_init_objectstoragepath(self): path = ObjectStoragePath("file://bucket/key/part1/part2") assert path.bucket == "bucket" - assert path.key == "key/part1/part2" - assert path._protocol == "file" + assert path.key == "/key/part1/part2" + assert path.protocol == "file" path2 = ObjectStoragePath(path / "part3") assert path2.bucket == "bucket" - assert path2.key == "key/part1/part2/part3" - assert path2._protocol == "file" + assert path2.key == "/key/part1/part2/part3" + assert path2.protocol == "file" - # check if we can append a non string to the path - path3 = ObjectStoragePath(path2 / 2023) + path3 = ObjectStoragePath(path2 / "2023") assert path3.bucket == "bucket" - assert path3.key == "key/part1/part2/part3/2023" - assert path3._protocol == "file" + assert path3.key == "/key/part1/part2/part3/2023" + assert path3.protocol == "file" def test_read_write(self): o = ObjectStoragePath(f"file:///tmp/{str(uuid.uuid4())}") @@ -87,69 +86,27 @@ def test_ls(self): filename = str(uuid.uuid4()) d = ObjectStoragePath(f"file:///tmp/{dirname}") - d.mkdir(create_parents=True) + d.mkdir(parents=True) o = d / filename o.touch() - data = d.ls() + data = list(d.iterdir()) assert len(data) == 1 - assert data[0]["name"] == o + assert data[0] == o - data = d.ls(detail=False) - assert data == [o] - - d.unlink(recursive=True) + d.rmdir(recursive=True) assert not o.exists() - def test_find(self): - dirname = str(uuid.uuid4()) - filename = str(uuid.uuid4()) - - d = ObjectStoragePath(f"file:///tmp/{dirname}") - d.mkdir(create_parents=True) - o = d / filename - o.touch() - - data = d.find("") - assert len(data) == 1 - assert data == [o] - - data = d.ls(detail=True) - assert len(data) == 1 - assert data[0]["name"] == o - - d.unlink(recursive=True) - @pytest.mark.parametrize( "fn, args, fn2, path, expected_args, expected_kwargs", [ - ("du", {}, "du", FOO, BAR, {"total": True, "maxdepth": None, "withdirs": False}), - ("exists", {}, "exists", FOO, ObjectStoragePath(BAR), {}), - ("checksum", {}, "checksum", FOO, ObjectStoragePath(BAR), {}), - ("size", {}, "size", FOO, ObjectStoragePath(BAR), {}), - ("is_dir", {}, "isdir", FOO, ObjectStoragePath(BAR), {}), - ("is_file", {}, "isfile", FOO, ObjectStoragePath(BAR), {}), - # ("is_symlink", {}, "islink", FOO, ObjectStoragePath(BAR), {}), - ("touch", {}, "touch", FOO, BAR, {"truncate": True}), - ("mkdir", {"exists_ok": True}, "mkdir", FOO, BAR, {"create_parents": True}), - ("read_text", {}, "read_text", FOO, BAR, {"encoding": None, "errors": None, "newline": None}), - ("read_bytes", {}, "read_bytes", FOO, BAR, {"start": None, "end": None}), - ("rm", {}, "rm", FOO, BAR, {"maxdepth": None, "recursive": False}), - ("rmdir", {}, "rmdir", FOO, BAR, {}), - ("write_bytes", {"data": b"foo"}, "pipe_file", FOO, ObjectStoragePath(BAR), {"value": b"foo"}), - ( - "write_text", - {"data": "foo"}, - "write_text", - FOO, - BAR, - {"value": "foo", "encoding": None, "errors": None, "newline": None}, - ), - ("ukey", {}, "ukey", FOO, BAR, {}), + ("checksum", {}, "checksum", FOO, FakeRemoteFileSystem._strip_protocol(BAR), {}), + ("size", {}, "size", FOO, FakeRemoteFileSystem._strip_protocol(BAR), {}), + ("ukey", {}, "ukey", FOO, FakeRemoteFileSystem._strip_protocol(BAR), {}), ], ) - def test_standard_api(self, fn, args, fn2, path, expected_args, expected_kwargs): + def test_standard_extended_api(self, fn, args, fn2, path, expected_args, expected_kwargs): _fs = mock.Mock() _fs._strip_protocol.return_value = "/" _fs.conn_id = "fake" @@ -210,16 +167,18 @@ def test_copy_remote_remote(self): assert (_to / _from.key / key).exists() assert (_to / _from.key / key).is_file() - _from.unlink(recursive=True) - _to.unlink(recursive=True) + _from.rmdir(recursive=True) + _to.rmdir(recursive=True) def test_serde_objectstoragepath(self): path = "file://bucket/key/part1/part2" o = ObjectStoragePath(path) + s = o.serialize() + assert s["path"] == path + d = ObjectStoragePath.deserialize(s, 1) - assert s["path"] == path assert o == d def test_serde_store(self): From 9f722123439ea922732e38298f781246086dd3f5 Mon Sep 17 00:00:00 2001 From: Bolke de Bruin Date: Mon, 13 Nov 2023 19:18:58 +0100 Subject: [PATCH 02/12] Add universal_pathlib --- setup.cfg | 1 + 1 file changed, 1 insertion(+) diff --git a/setup.cfg b/setup.cfg index 2cda0924cc533..1bbf5fb8f1a70 100644 --- a/setup.cfg +++ b/setup.cfg @@ -163,6 +163,7 @@ install_requires = # See https://github.com/apache/airflow/pull/31693 # We should also remove "licenses/LICENSE-unicodecsv.txt" file when we remove this dependency unicodecsv>=0.14.1 + universal_pathlib>=0.1.4 werkzeug>=2.0 [options.packages.find] From 5a415574cecd6497903b538809d23666d11c14ee Mon Sep 17 00:00:00 2001 From: Bolke de Bruin Date: Mon, 13 Nov 2023 20:12:26 +0100 Subject: [PATCH 03/12] Moving stuff to utils --- airflow/io/store/path.py | 2 +- airflow/io/utils/__init__.py | 16 ++++++++++++++++ airflow/io/{store => utils}/stat.py | 0 3 files changed, 17 insertions(+), 1 deletion(-) create mode 100644 airflow/io/utils/__init__.py rename airflow/io/{store => utils}/stat.py (100%) diff --git a/airflow/io/store/path.py b/airflow/io/store/path.py index 77e74d3b7422c..a558d58119df4 100644 --- a/airflow/io/store/path.py +++ b/airflow/io/store/path.py @@ -32,7 +32,7 @@ from airflow.io.registry import get_path_class from airflow.io.store import attach -from airflow.io.store.stat import stat_result +from airflow.io.utils.stat import stat_result if typing.TYPE_CHECKING: from fsspec import AbstractFileSystem diff --git a/airflow/io/utils/__init__.py b/airflow/io/utils/__init__.py new file mode 100644 index 0000000000000..13a83393a9124 --- /dev/null +++ b/airflow/io/utils/__init__.py @@ -0,0 +1,16 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. diff --git a/airflow/io/store/stat.py b/airflow/io/utils/stat.py similarity index 100% rename from airflow/io/store/stat.py rename to airflow/io/utils/stat.py From 40b34a7272354d5aad70c30b122d9ae4d4707d49 Mon Sep 17 00:00:00 2001 From: Bolke de Bruin Date: Mon, 13 Nov 2023 20:25:24 +0100 Subject: [PATCH 04/12] Refactor ObjectStoragePath to airflow.io.path --- airflow/example_dags/tutorial_objectstorage.py | 2 +- airflow/io/implementations.py | 2 +- airflow/io/{store => }/path.py | 0 airflow/io/registry.py | 8 ++++---- airflow/providers/common/io/operators/file_transfer.py | 2 +- docs/apache-airflow/core-concepts/objectstorage.rst | 8 ++++---- tests/io/store/test_store.py | 2 +- .../common/io/example_file_transfer_local_to_s3.py | 2 +- 8 files changed, 13 insertions(+), 13 deletions(-) rename airflow/io/{store => }/path.py (100%) diff --git a/airflow/example_dags/tutorial_objectstorage.py b/airflow/example_dags/tutorial_objectstorage.py index 5394238c7ad2b..e3a2c32705d73 100644 --- a/airflow/example_dags/tutorial_objectstorage.py +++ b/airflow/example_dags/tutorial_objectstorage.py @@ -23,7 +23,7 @@ import requests from airflow.decorators import dag, task -from airflow.io.store.path import ObjectStoragePath +from airflow.io.path import ObjectStoragePath # [END import_module] diff --git a/airflow/io/implementations.py b/airflow/io/implementations.py index db492ad52b4a4..e91b4587495ee 100644 --- a/airflow/io/implementations.py +++ b/airflow/io/implementations.py @@ -19,7 +19,7 @@ from upath.implementations.cloud import CloudPath -from airflow.io.store.path import ObjectStoragePath, _AirflowCloudAccessor +from airflow.io.path import ObjectStoragePath, _AirflowCloudAccessor class AirflowCloudPath(ObjectStoragePath, CloudPath): diff --git a/airflow/io/store/path.py b/airflow/io/path.py similarity index 100% rename from airflow/io/store/path.py rename to airflow/io/path.py diff --git a/airflow/io/registry.py b/airflow/io/registry.py index 7cbf94d6152e8..063af1b2478af 100644 --- a/airflow/io/registry.py +++ b/airflow/io/registry.py @@ -27,12 +27,12 @@ import upath if TYPE_CHECKING: - import airflow.io.store.path + import airflow.io.path _ENTRY_POINT_GROUP = "airflow_pathlib.implementations" -class _Registry(MutableMapping[str, "type[airflow.io.store.path.ObjectStoragePath]"]): +class _Registry(MutableMapping[str, "type[airflow.io.path.ObjectStoragePath]"]): """Registry for ObjectStoragePath subclasses.""" # _default_ set for serialization @@ -46,7 +46,7 @@ def __init__(self) -> None: self._entries = {ep.name: ep for ep in eps} self._m = ChainMap({}, self.known_implementations) # type: ignore - def __getitem__(self, item: str) -> type[airflow.io.store.path.ObjectStoragePath]: + def __getitem__(self, item: str) -> type[airflow.io.path.ObjectStoragePath]: fqn = self._m.get(item) if fqn is None: if item in self._entries: @@ -82,7 +82,7 @@ def __iter__(self): @lru_cache def get_path_class( protocol: str, *, default: str | None = None -) -> type[upath.core.UPath | airflow.io.store.path.ObjectStoragePath]: +) -> type[upath.core.UPath | airflow.io.path.ObjectStoragePath]: """Return the path cls for the given protocol. Returns `None` or ``default`` if no matching protocol can be found. diff --git a/airflow/providers/common/io/operators/file_transfer.py b/airflow/providers/common/io/operators/file_transfer.py index f23fa422d9ef7..e720f786669f0 100644 --- a/airflow/providers/common/io/operators/file_transfer.py +++ b/airflow/providers/common/io/operators/file_transfer.py @@ -19,7 +19,7 @@ from typing import TYPE_CHECKING, Sequence -from airflow.io.store.path import ObjectStoragePath +from airflow.io.path import ObjectStoragePath from airflow.models import BaseOperator if TYPE_CHECKING: diff --git a/docs/apache-airflow/core-concepts/objectstorage.rst b/docs/apache-airflow/core-concepts/objectstorage.rst index 194db0477e154..25fa29555d35a 100644 --- a/docs/apache-airflow/core-concepts/objectstorage.rst +++ b/docs/apache-airflow/core-concepts/objectstorage.rst @@ -70,7 +70,7 @@ object you want to interact with. For example, to point to a bucket in s3, you w .. code-block:: python - from airflow.io.store.path import ObjectStoragePath + from airflow.io.path import ObjectStoragePath base = ObjectStoragePath("s3://my-bucket/", conn_id="aws_default") # conn_id is optional @@ -149,7 +149,7 @@ would do the following: .. code-block:: python - from airflow.io.store.path import ObjectStoragePath + from airflow.io.path import ObjectStoragePath from airflow.io.store import attach from fsspec.implementations.dbfs import DBFSFileSystem @@ -172,7 +172,7 @@ The object storage abstraction is implemented as a `Path-like API Date: Mon, 13 Nov 2023 20:42:40 +0100 Subject: [PATCH 05/12] Remove redundant parts --- airflow/io/implementations.py | 28 --------- airflow/io/path.py | 10 +--- airflow/io/registry.py | 103 ---------------------------------- 3 files changed, 3 insertions(+), 138 deletions(-) delete mode 100644 airflow/io/implementations.py delete mode 100644 airflow/io/registry.py diff --git a/airflow/io/implementations.py b/airflow/io/implementations.py deleted file mode 100644 index e91b4587495ee..0000000000000 --- a/airflow/io/implementations.py +++ /dev/null @@ -1,28 +0,0 @@ -# -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. -from __future__ import annotations - -from upath.implementations.cloud import CloudPath - -from airflow.io.path import ObjectStoragePath, _AirflowCloudAccessor - - -class AirflowCloudPath(ObjectStoragePath, CloudPath): - """A Path implementation for that uses Airflow's Connections.""" - - _default_accessor = _AirflowCloudAccessor diff --git a/airflow/io/path.py b/airflow/io/path.py index a558d58119df4..1b1b0e719089b 100644 --- a/airflow/io/path.py +++ b/airflow/io/path.py @@ -26,11 +26,9 @@ from fsspec.core import split_protocol from fsspec.utils import stringify_path -from upath import UPath -from upath.implementations.cloud import _CloudAccessor +from upath.implementations.cloud import CloudPath, _CloudAccessor from upath.registry import get_upath_class -from airflow.io.registry import get_path_class from airflow.io.store import attach from airflow.io.utils.stat import stat_result @@ -65,7 +63,7 @@ def __eq__(self, other): return isinstance(other, _AirflowCloudAccessor) and self._store == other._store -class ObjectStoragePath(UPath): +class ObjectStoragePath(CloudPath): """A path-like object for object storage.""" __version__: typing.ClassVar[int] = 1 @@ -128,8 +126,6 @@ def __new__(cls: type[PT], *args: str | os.PathLike, **kwargs: typing.Any) -> PT else: protocol = kwargs.get("scheme", protocol) - path_cls = get_path_class(protocol, default="_default_") - for key in ["scheme", "netloc"]: val = kwargs.get(key) if val: @@ -143,7 +139,7 @@ def __new__(cls: type[PT], *args: str | os.PathLike, **kwargs: typing.Any) -> PT else: args_list.insert(0, parsed_url.path) - return path_cls._from_parts(args_list, url=parsed_url, **kwargs) # type: ignore + return cls._from_parts(args_list, url=parsed_url, **kwargs) # type: ignore @functools.lru_cache def __hash__(self) -> int: diff --git a/airflow/io/registry.py b/airflow/io/registry.py deleted file mode 100644 index 063af1b2478af..0000000000000 --- a/airflow/io/registry.py +++ /dev/null @@ -1,103 +0,0 @@ -# -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. -from __future__ import annotations - -import sys -from collections import ChainMap -from functools import lru_cache -from importlib import import_module -from importlib.metadata import entry_points -from typing import TYPE_CHECKING, MutableMapping - -import upath - -if TYPE_CHECKING: - import airflow.io.path - -_ENTRY_POINT_GROUP = "airflow_pathlib.implementations" - - -class _Registry(MutableMapping[str, "type[airflow.io.path.ObjectStoragePath]"]): - """Registry for ObjectStoragePath subclasses.""" - - # _default_ set for serialization - known_implementations: dict[str, str] = {"_default_": "airflow.io.implementations.AirflowCloudPath"} - - def __init__(self) -> None: - if sys.version_info >= (3, 10): - eps = entry_points(group=_ENTRY_POINT_GROUP) - else: - eps = entry_points().get(_ENTRY_POINT_GROUP, []) - self._entries = {ep.name: ep for ep in eps} - self._m = ChainMap({}, self.known_implementations) # type: ignore - - def __getitem__(self, item: str) -> type[airflow.io.path.ObjectStoragePath]: - fqn = self._m.get(item) - if fqn is None: - if item in self._entries: - fqn = self._m[item] = self._entries[item].load() - if fqn is None: - raise KeyError(f"{item} not in registry") - if isinstance(fqn, str): - module_name, name = fqn.rsplit(".", 1) - mod = import_module(module_name) - cls = getattr(mod, name) # type: ignore - else: - cls = fqn - return cls - - def __setitem__(self, item: str, value: type[upath.core.UPath] | str) -> None: - if not ((isinstance(value, type) and issubclass(value, upath.core.UPath)) or isinstance(value, str)): - raise ValueError(f"expected ObjectStorage subclass or FQN-string, got: {type(value).__name__!r}") - self._m[item] = value - - def __delitem__(self, __v: str) -> None: - raise NotImplementedError("removal is unsupported") - - def __len__(self) -> int: - return len(set().union(self._m, self._entries)) - - def __iter__(self): - return iter(set().union(self._m, self._entries)) - - -registry = _Registry() - - -@lru_cache -def get_path_class( - protocol: str, *, default: str | None = None -) -> type[upath.core.UPath | airflow.io.path.ObjectStoragePath]: - """Return the path cls for the given protocol. - - Returns `None` or ``default`` if no matching protocol can be found. - - Parameters - ---------- - protocol: - The protocol string - default: - returned if no matching protocol can be found - """ - try: - return registry[protocol] - except KeyError: - if default: - return registry[default] - else: - raise From e804bc4dd3b21ddcc97ab4aebd089b15ad4d25de Mon Sep 17 00:00:00 2001 From: Bolke de Bruin Date: Tue, 14 Nov 2023 10:56:46 +0100 Subject: [PATCH 06/12] Update docs --- .../core-concepts/objectstorage.rst | 115 +++++++++++------- 1 file changed, 70 insertions(+), 45 deletions(-) diff --git a/docs/apache-airflow/core-concepts/objectstorage.rst b/docs/apache-airflow/core-concepts/objectstorage.rst index 25fa29555d35a..3fd7b57e72573 100644 --- a/docs/apache-airflow/core-concepts/objectstorage.rst +++ b/docs/apache-airflow/core-concepts/objectstorage.rst @@ -65,7 +65,7 @@ However, you should be aware of the limitations of object storage when designing Basic Use --------- -To use object storage, you need to instantiate a Path-like (see below) object with the URI of the +To use object storage, you need to instantiate a Path (see below) object with the URI of the object you want to interact with. For example, to point to a bucket in s3, you would do the following: .. code-block:: python @@ -165,27 +165,15 @@ would do the following: .. _concepts:api: -Path-like API +Path API ------------- -The object storage abstraction is implemented as a `Path-like API `_. -This means that you can mostly use the same API to interact with object storage as you would with a local filesystem. -In this section we only list the differences between the two APIs. Extended operations beyond the standard Path API -, like copying and moving, are listed in the next section. For details about each operation, like what arguments -they take, see the documentation of the :class:`~airflow.io.path.ObjectStoragePath` class. - - -stat -^^^^ - -Returns a ``stat_result`` like object that supports the following attributes: ``st_size``, ``st_mtime``, ``st_mode``, -but also acts like a dictionary that can provide additional metadata about the object. For example, for s3 it will, -return the additional keys like: ``['ETag', 'ContentType']``. If your code needs to be portable across different object -store do not rely on the extended metadata. - -.. note:: - While ``stat`` does accept the ``follow_symlinks`` argument, it is not passed on to the object storage backend as - not all object storage does not support symlinks. +The object storage abstraction is implemented as a `Path API `_. +and builds upon `Universal Pathlib `_ This means that you can mostly use +the same API to interact with object storage as you would with a local filesystem. In this section we only list the +differences between the two APIs. Extended operations beyond the standard Path API, like copying and moving, are listed +in the next section. For details about each operation, like what arguments they take, see the documentation of +the :class:`~airflow.io.path.ObjectStoragePath` class. mkdir @@ -194,57 +182,89 @@ mkdir Create a directory entry at the specified path or within a bucket/container. For systems that don't have true directories, it may create a directory entry for this instance only and not affect the real filesystem. -If ``create_parents`` is ``True`` (the default), any missing parents of this path are created as needed. +If ``parents`` is ``True``, any missing parents of this path are created as needed. touch ^^^^^ -Create an empty file, or update the timestamp. If ``truncate`` is ``True``, the file is truncated, which is the -default. +Create a file at this given path, or update the timestamp. If ``truncate`` is ``True``, the file is truncated, which is +the default. If the file already exists, the function succeeds if ``exists_ok`` is true (and its modification time is +updated to the current time), otherwise ``FileExistsError`` is raised. + + +stat +^^^^ + +Returns a ``stat_result`` like object that supports the following attributes: ``st_size``, ``st_mtime``, ``st_mode``, +but also acts like a dictionary that can provide additional metadata about the object. For example, for s3 it will, +return the additional keys like: ``['ETag', 'ContentType']``. If your code needs to be portable across different object +stores do not rely on the extended metadata. + +.. note:: + While ``stat`` does accept the ``follow_symlinks`` argument, it is not passed on to the object storage backend as + not all object storage does not support symlinks. .. _concepts:extended-operations: -Extended Operations -------------------- +Extensions +---------- The following operations are not part of the standard Path API, but are supported by the object storage abstraction. -ukey -^^^^ +bucket +^^^^^^ -Hash of file properties, to tell if it has changed. +Returns the bucket name. checksum ^^^^^^^^ -Return the checksum of the file. +Returns the checksum of the file. -read_block -^^^^^^^^^^ +container +^^^^^^^^^ -Read a block of bytes from the file. This is useful for reading large files in chunks. +Alias of bucket -du +fs ^^ -Space used by files and optionally directories within a path. +Convenience attribute to access an instantiated filesystem -find +key +^^^ + +Returns the object key. + + +path ^^^^ +the ``fsspec`` compatible path for use with filesystem instances -Find files and optionally directories within a path. +protocol +^^^^^^^^ -ls -^^ +the filesystem_spec protocol. -List files within a path. + +read_block +^^^^^^^^^^ + +Read a block of bytes from the file at this given path. + +Starting at offset of the file, read length bytes. If delimiter is set then we ensure +that the read starts and stops at delimiter boundaries that follow the locations offset +and offset + length. If offset is zero then we start at zero. The bytestring returned +WILL include the end delimiter string. + +If offset+length is beyond the eof, reads to eof. sign @@ -254,17 +274,22 @@ Create a signed URL representing the given path. Some implementations allow temp way of delegating credentials. -copy +size ^^^^ -Copy a file from one path to another. If the destination is a directory, the file will be copied into it. If the -destination is a file, it will be overwritten. +Returns the size in bytes of the file at the given path. + -move +storage_options +^^^^^^^^^^^^^^^ + +The storage options for instantiating the underlying filesystem. + + +ukey ^^^^ -Move a file from one path to another. If the destination is a directory, the file will be moved into it. If the -destination is a file, it will be overwritten. +Hash of file properties, to tell if it has changed. .. _concepts:copying-and-moving: From 22fb3819aa0f93a5aceeee47286214a0a0c3f8c1 Mon Sep 17 00:00:00 2001 From: Bolke de Bruin Date: Tue, 14 Nov 2023 12:57:00 +0100 Subject: [PATCH 07/12] Fix mypy and docs --- .../example_dags/tutorial_objectstorage.py | 2 +- airflow/io/path.py | 21 +++++++++++++------ .../core-concepts/objectstorage.rst | 4 ---- docs/spelling_wordlist.txt | 1 + .../io/example_file_transfer_local_to_s3.py | 2 +- 5 files changed, 18 insertions(+), 12 deletions(-) diff --git a/airflow/example_dags/tutorial_objectstorage.py b/airflow/example_dags/tutorial_objectstorage.py index e3a2c32705d73..47db595c24475 100644 --- a/airflow/example_dags/tutorial_objectstorage.py +++ b/airflow/example_dags/tutorial_objectstorage.py @@ -93,7 +93,7 @@ def get_air_quality_data(**kwargs) -> ObjectStoragePath: response.raise_for_status() # ensure the bucket exists - base.mkdir(exists_ok=True) + base.mkdir(exist_ok=True) formatted_date = execution_date.format("YYYYMMDD") path = base / f"air_quality_{formatted_date}.parquet" diff --git a/airflow/io/path.py b/airflow/io/path.py index 1b1b0e719089b..e305614f5613f 100644 --- a/airflow/io/path.py +++ b/airflow/io/path.py @@ -29,7 +29,7 @@ from upath.implementations.cloud import CloudPath, _CloudAccessor from upath.registry import get_upath_class -from airflow.io.store import attach +from airflow.io.store import ObjectStore, attach from airflow.io.utils.stat import stat_result if typing.TYPE_CHECKING: @@ -42,6 +42,9 @@ class _AirflowCloudAccessor(_CloudAccessor): + _store: ObjectStore + _conn_id: str | None + __slots__ = ("_store", "_conn_id") def __init__(self, parsed_url: SplitResult | None, **kwargs: typing.Any) -> None: @@ -154,15 +157,21 @@ def container(self) -> str: @property def bucket(self) -> str: - return self._url.netloc + if self._url: + return self._url.netloc + else: + return "" @property def key(self) -> str: - return self._url.path + if self._url: + return self._url.path + else: + return "" - def stat(self, *, follow_symlinks: bool = True) -> stat_result: + def stat(self) -> stat_result: """Return the result of the `stat()` call.""" # noqa: D402 - return stat_result(self._accessor.stat(self), protocol=self.protocol, conn_id=self._accessor.conn_id) + return stat_result(self._accessor.stat(self), protocol=self.protocol, conn_id=self._accessor._conn_id) def samefile(self, other_path: typing.Any) -> bool: """Return whether other_path is the same or not as this file.""" @@ -183,7 +192,7 @@ def _scandir(self): # context manager. return contextlib.nullcontext(self.iterdir()) - def replace(self, target: str | ObjectStoragePath) -> ObjectStoragePath: + def replace(self, target) -> ObjectStoragePath: """ Rename this path to the target path, overwriting if that path exists. diff --git a/docs/apache-airflow/core-concepts/objectstorage.rst b/docs/apache-airflow/core-concepts/objectstorage.rst index 3fd7b57e72573..20418d84ff8a9 100644 --- a/docs/apache-airflow/core-concepts/objectstorage.rst +++ b/docs/apache-airflow/core-concepts/objectstorage.rst @@ -201,10 +201,6 @@ but also acts like a dictionary that can provide additional metadata about the o return the additional keys like: ``['ETag', 'ContentType']``. If your code needs to be portable across different object stores do not rely on the extended metadata. -.. note:: - While ``stat`` does accept the ``follow_symlinks`` argument, it is not passed on to the object storage backend as - not all object storage does not support symlinks. - .. _concepts:extended-operations: diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt index 3979ae3c597e6..a7e864f37ca06 100644 --- a/docs/spelling_wordlist.txt +++ b/docs/spelling_wordlist.txt @@ -548,6 +548,7 @@ env envFrom EnvVar envvar +eof eol eols eRevalue diff --git a/tests/system/providers/common/io/example_file_transfer_local_to_s3.py b/tests/system/providers/common/io/example_file_transfer_local_to_s3.py index 8087ea6e4cf8d..13c495e620972 100644 --- a/tests/system/providers/common/io/example_file_transfer_local_to_s3.py +++ b/tests/system/providers/common/io/example_file_transfer_local_to_s3.py @@ -57,7 +57,7 @@ def delete_temp_file(path: ObjectStoragePath): @task def remove_bucket(): - AWS_BUCKET.unlink(recursive=True) + AWS_BUCKET.rmdir(recursive=True) with DAG( From f7426899c228709e98cc9958130e8a1a79a045b3 Mon Sep 17 00:00:00 2001 From: Bolke de Bruin Date: Tue, 14 Nov 2023 15:05:06 +0100 Subject: [PATCH 08/12] Fix errors add tests --- airflow/io/path.py | 2 +- airflow/io/utils/stat.py | 10 ++-- tests/io/store/__init__.py | 16 ----- .../io/{store/test_store.py => test_path.py} | 58 +++++++++++++++++++ 4 files changed, 64 insertions(+), 22 deletions(-) delete mode 100644 tests/io/store/__init__.py rename tests/io/{store/test_store.py => test_path.py} (78%) diff --git a/airflow/io/path.py b/airflow/io/path.py index e305614f5613f..b33871e9f46cc 100644 --- a/airflow/io/path.py +++ b/airflow/io/path.py @@ -262,7 +262,7 @@ def read_block(self, offset: int, length: int, delimiter=None): -------- :func:`fsspec.utils.read_block` """ - return self.fs.read_block(self.path, offset, length, delimiter=delimiter) + return self.fs.read_block(self.path, offset=offset, length=length, delimiter=delimiter) def sign(self, expiration: int = 100, **kwargs): """Create a signed URL representing the given path. diff --git a/airflow/io/utils/stat.py b/airflow/io/utils/stat.py index 75131cda95987..6d873b614df41 100644 --- a/airflow/io/utils/stat.py +++ b/airflow/io/utils/stat.py @@ -36,19 +36,19 @@ class stat_result(dict): st_dev = property(lambda self: 0) """device""" - st_size = property(lambda self: self._info.get("size", 0)) + st_size = property(lambda self: self.get("size", 0)) """total size, in bytes""" - st_gid = property(lambda self: self._info.get("gid", 0)) + st_gid = property(lambda self: self.get("gid", 0)) """group ID of owner""" - st_uid = property(lambda self: self._info.get("uid", 0)) + st_uid = property(lambda self: self.get("uid", 0)) """user ID of owner""" - st_ino = property(lambda self: self._info.get("ino", 0)) + st_ino = property(lambda self: self.get("ino", 0)) """inode""" - st_nlink = property(lambda self: self._info.get("nlink", 0)) + st_nlink = property(lambda self: self.get("nlink", 0)) """number of hard links""" @property diff --git a/tests/io/store/__init__.py b/tests/io/store/__init__.py deleted file mode 100644 index 13a83393a9124..0000000000000 --- a/tests/io/store/__init__.py +++ /dev/null @@ -1,16 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. diff --git a/tests/io/store/test_store.py b/tests/io/test_path.py similarity index 78% rename from tests/io/store/test_store.py rename to tests/io/test_path.py index ffb7fc9c632ea..488ff8df06246 100644 --- a/tests/io/store/test_store.py +++ b/tests/io/test_path.py @@ -18,6 +18,8 @@ from __future__ import annotations import uuid +from stat import S_ISDIR, S_ISREG +from tempfile import NamedTemporaryFile from unittest import mock import pytest @@ -103,7 +105,23 @@ def test_ls(self): [ ("checksum", {}, "checksum", FOO, FakeRemoteFileSystem._strip_protocol(BAR), {}), ("size", {}, "size", FOO, FakeRemoteFileSystem._strip_protocol(BAR), {}), + ( + "sign", + {"expiration": 200, "extra": "xtra"}, + "sign", + FOO, + FakeRemoteFileSystem._strip_protocol(BAR), + {"expiration": 200, "extra": "xtra"}, + ), ("ukey", {}, "ukey", FOO, FakeRemoteFileSystem._strip_protocol(BAR), {}), + ( + "read_block", + {"offset": 0, "length": 1}, + "read_block", + FOO, + FakeRemoteFileSystem._strip_protocol(BAR), + {"delimiter": None, "length": 1, "offset": 0}, + ), ], ) def test_standard_extended_api(self, fn, args, fn2, path, expected_args, expected_kwargs): @@ -117,6 +135,46 @@ def test_standard_extended_api(self, fn, args, fn2, path, expected_args, expecte getattr(o, fn)(**args) getattr(store.fs, fn2).assert_called_once_with(expected_args, **expected_kwargs) + def test_stat(self): + with NamedTemporaryFile() as f: + o = ObjectStoragePath(f"file://{f.name}") + assert o.stat().st_size == 0 + assert S_ISREG(o.stat().st_mode) + assert S_ISDIR(o.parent.stat().st_mode) + + def test_bucket_key_protocol(self): + bucket = "bkt" + key = "yek" + protocol = "s3" + + o = ObjectStoragePath(f"{protocol}://{bucket}/{key}") + assert o.bucket == bucket + assert o.container == bucket + assert o.key == f"/{key}" + assert o.protocol == protocol + + def test_cwd_home(self): + assert ObjectStoragePath.cwd() + assert ObjectStoragePath.home() + + def test_replace(self): + o = ObjectStoragePath(f"file:///tmp/{str(uuid.uuid4())}") + i = ObjectStoragePath(f"file:///tmp/{str(uuid.uuid4())}") + + o.touch() + i.touch() + + assert i.size() == 0 + + txt = "foo" + o.write_text(txt) + e = o.replace(i) + assert o.exists() is False + assert i == e + assert e.size() == len(txt) + + e.unlink() + def test_move_local(self): _from = ObjectStoragePath(f"file:///tmp/{str(uuid.uuid4())}") _to = ObjectStoragePath(f"file:///tmp/{str(uuid.uuid4())}") From d5d97b24fa0ba1b7583184cb26233aada77baf6b Mon Sep 17 00:00:00 2001 From: Bolke de Bruin Date: Tue, 14 Nov 2023 16:06:23 +0100 Subject: [PATCH 09/12] Some mypy --- airflow/io/path.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/airflow/io/path.py b/airflow/io/path.py index b33871e9f46cc..c265bed21639c 100644 --- a/airflow/io/path.py +++ b/airflow/io/path.py @@ -169,9 +169,9 @@ def key(self) -> str: else: return "" - def stat(self) -> stat_result: + def stat(self) -> stat_result: # type: ignore[override] """Return the result of the `stat()` call.""" # noqa: D402 - return stat_result(self._accessor.stat(self), protocol=self.protocol, conn_id=self._accessor._conn_id) + return stat_result(self._accessor.stat(self), protocol=self.protocol, conn_id=self._accessor._conn_id) # type: ignore[attr-defined] def samefile(self, other_path: typing.Any) -> bool: """Return whether other_path is the same or not as this file.""" @@ -347,7 +347,7 @@ def copy(self, dst: str | ObjectStoragePath, recursive: bool = False, **kwargs) if path == self.path: continue - src_obj = ObjectStoragePath(path, store=self._accessor._store) + src_obj = ObjectStoragePath(path, store=self._accessor._store) # type: ignore[attr-defined] # skip directories, empty directories will not be created if src_obj.is_dir(): From 8048b6472049223b2d623062c9a7ec1a20a9bf73 Mon Sep 17 00:00:00 2001 From: Tzu-ping Chung Date: Wed, 15 Nov 2023 16:14:37 +0800 Subject: [PATCH 10/12] Simplify path accessor Since attach() already caches store objects, I feel we shouldn't need to store all the extra information. We can simply take conn_id on Path, and rely on the internal caching to retrieve the correct information afterwards. --- airflow/io/path.py | 62 +++++++++++++++++++--------------------------- 1 file changed, 25 insertions(+), 37 deletions(-) diff --git a/airflow/io/path.py b/airflow/io/path.py index c265bed21639c..f87bdfbbcadad 100644 --- a/airflow/io/path.py +++ b/airflow/io/path.py @@ -22,7 +22,7 @@ import shutil import typing from pathlib import PurePath -from urllib.parse import SplitResult, urlsplit +from urllib.parse import urlsplit from fsspec.core import split_protocol from fsspec.utils import stringify_path @@ -33,6 +33,8 @@ from airflow.io.utils.stat import stat_result if typing.TYPE_CHECKING: + from urllib.parse import SplitResult + from fsspec import AbstractFileSystem @@ -42,21 +44,10 @@ class _AirflowCloudAccessor(_CloudAccessor): - _store: ObjectStore - _conn_id: str | None - - __slots__ = ("_store", "_conn_id") - - def __init__(self, parsed_url: SplitResult | None, **kwargs: typing.Any) -> None: - _store = kwargs.pop("store", None) - conn_id = kwargs.pop("conn_id", None) - if _store: - self._store = _store - elif parsed_url and parsed_url.scheme: - self._store = attach(parsed_url.scheme, conn_id) # todo add kwargs as storage_options - else: - self._store = attach(default, conn_id) - self._conn_id = conn_id + __slots__ = ("_store",) + + def __init__(self, *, store: ObjectStore) -> None: + self._store = store @property def _fs(self) -> AbstractFileSystem: @@ -69,6 +60,8 @@ def __eq__(self, other): class ObjectStoragePath(CloudPath): """A path-like object for object storage.""" + _accessor: _AirflowCloudAccessor + __version__: typing.ClassVar[int] = 1 _default_accessor = _AirflowCloudAccessor @@ -89,7 +82,7 @@ class ObjectStoragePath(CloudPath): "_hash", ) - def __new__(cls: type[PT], *args: str | os.PathLike, **kwargs: typing.Any) -> PT: + def __new__(cls: type[PT], *args: str | os.PathLike, conn_id: str | None = None) -> PT: args_list = list(args) try: other = args_list.pop(0) @@ -99,7 +92,7 @@ def __new__(cls: type[PT], *args: str | os.PathLike, **kwargs: typing.Any) -> PT other = other or "." if isinstance(other, PurePath): - _cls: type[typing.Any] = type(other) + _cls: typing.Any = type(other) drv, root, parts = _cls._parse_args(args_list) drv, root, parts = _cls._flavour.join_parsed_parts( other._drv, # type: ignore[attr-defined] @@ -115,34 +108,25 @@ def __new__(cls: type[PT], *args: str | os.PathLike, **kwargs: typing.Any) -> PT other_kwargs = _kwargs.copy() if _url and _url.scheme: other_kwargs["url"] = _url - new_kwargs = _kwargs.copy() - new_kwargs.update(kwargs) + _kwargs["conn_id"] = conn_id - return _cls(_cls._format_parsed_parts(drv, root, parts, **other_kwargs), **new_kwargs) + return _cls(_cls._format_parsed_parts(drv, root, parts, **other_kwargs), **_kwargs) url = stringify_path(other) - protocol, _ = split_protocol(url) - parsed_url = urlsplit(url) - - if protocol is None: - protocol = kwargs.get("scheme", parsed_url.scheme) - else: - protocol = kwargs.get("scheme", protocol) - - for key in ["scheme", "netloc"]: - val = kwargs.get(key) - if val: - parsed_url = parsed_url._replace(**{key: val}) + parsed_url: SplitResult = urlsplit(url) + protocol: str | None = split_protocol(url)[0] or parsed_url.scheme if not parsed_url.path: parsed_url = parsed_url._replace(path="/") # ensure path has root if not protocol: args_list.insert(0, url) + store = attach(default, conn_id) else: args_list.insert(0, parsed_url.path) + store = attach(protocol, conn_id) - return cls._from_parts(args_list, url=parsed_url, **kwargs) # type: ignore + return cls._from_parts(args_list, url=parsed_url, store=store) @functools.lru_cache def __hash__(self) -> int: @@ -170,8 +154,12 @@ def key(self) -> str: return "" def stat(self) -> stat_result: # type: ignore[override] - """Return the result of the `stat()` call.""" # noqa: D402 - return stat_result(self._accessor.stat(self), protocol=self.protocol, conn_id=self._accessor._conn_id) # type: ignore[attr-defined] + """Call ``stat`` and return the result.""" + return stat_result( + self._accessor.stat(self), + protocol=self.protocol, + conn_id=self._accessor._store.conn_id, + ) def samefile(self, other_path: typing.Any) -> bool: """Return whether other_path is the same or not as this file.""" @@ -347,7 +335,7 @@ def copy(self, dst: str | ObjectStoragePath, recursive: bool = False, **kwargs) if path == self.path: continue - src_obj = ObjectStoragePath(path, store=self._accessor._store) # type: ignore[attr-defined] + src_obj = ObjectStoragePath(path, conn_id=self._accessor._store.conn_id) # skip directories, empty directories will not be created if src_obj.is_dir(): From 24450e0c56e2f48e290bd4dd5200d5e377ef6c93 Mon Sep 17 00:00:00 2001 From: Bolke de Bruin Date: Fri, 17 Nov 2023 13:42:29 +0100 Subject: [PATCH 11/12] Remove need for storing conn_id add experimental --- airflow/io/path.py | 33 ++++++++++++++----- .../core-concepts/objectstorage.rst | 2 ++ tests/io/test_path.py | 11 +++++-- 3 files changed, 35 insertions(+), 11 deletions(-) diff --git a/airflow/io/path.py b/airflow/io/path.py index f87bdfbbcadad..f308096f67444 100644 --- a/airflow/io/path.py +++ b/airflow/io/path.py @@ -29,7 +29,7 @@ from upath.implementations.cloud import CloudPath, _CloudAccessor from upath.registry import get_upath_class -from airflow.io.store import ObjectStore, attach +from airflow.io.store import attach from airflow.io.utils.stat import stat_result if typing.TYPE_CHECKING: @@ -46,8 +46,15 @@ class _AirflowCloudAccessor(_CloudAccessor): __slots__ = ("_store",) - def __init__(self, *, store: ObjectStore) -> None: - self._store = store + def __init__(self, parsed_url: SplitResult | None, **kwargs: typing.Any) -> None: + store = kwargs.pop("store", None) + conn_id = kwargs.pop("conn_id", None) + if store: + self._store = store + elif parsed_url and parsed_url.scheme: + self._store = attach(parsed_url.scheme, conn_id) + else: + self._store = attach(default, conn_id) @property def _fs(self) -> AbstractFileSystem: @@ -82,8 +89,9 @@ class ObjectStoragePath(CloudPath): "_hash", ) - def __new__(cls: type[PT], *args: str | os.PathLike, conn_id: str | None = None) -> PT: + def __new__(cls: type[PT], *args: str | os.PathLike, **kwargs: typing.Any) -> PT: args_list = list(args) + try: other = args_list.pop(0) except IndexError: @@ -108,25 +116,32 @@ def __new__(cls: type[PT], *args: str | os.PathLike, conn_id: str | None = None) other_kwargs = _kwargs.copy() if _url and _url.scheme: other_kwargs["url"] = _url - _kwargs["conn_id"] = conn_id + new_kwargs = _kwargs.copy() + new_kwargs.update(kwargs) - return _cls(_cls._format_parsed_parts(drv, root, parts, **other_kwargs), **_kwargs) + return _cls(_cls._format_parsed_parts(drv, root, parts, **other_kwargs), **new_kwargs) url = stringify_path(other) parsed_url: SplitResult = urlsplit(url) protocol: str | None = split_protocol(url)[0] or parsed_url.scheme + # allow override of protocol + protocol = kwargs.get("scheme", protocol) + + for key in ["scheme", "url"]: + val = kwargs.pop(key, None) + if val: + parsed_url = parsed_url._replace(**{key: val}) + if not parsed_url.path: parsed_url = parsed_url._replace(path="/") # ensure path has root if not protocol: args_list.insert(0, url) - store = attach(default, conn_id) else: args_list.insert(0, parsed_url.path) - store = attach(protocol, conn_id) - return cls._from_parts(args_list, url=parsed_url, store=store) + return cls._from_parts(args_list, url=parsed_url, **kwargs) # type: ignore @functools.lru_cache def __hash__(self) -> int: diff --git a/docs/apache-airflow/core-concepts/objectstorage.rst b/docs/apache-airflow/core-concepts/objectstorage.rst index 20418d84ff8a9..046cb4852278c 100644 --- a/docs/apache-airflow/core-concepts/objectstorage.rst +++ b/docs/apache-airflow/core-concepts/objectstorage.rst @@ -23,6 +23,8 @@ Object Storage .. versionadded:: 2.8.0 +|experimental| + All major cloud providers offer persistent data storage in object stores. These are not classic "POSIX" file systems. In order to store hundreds of petabytes of data without any single points of failure, object stores replace the classic file system directory tree with a simpler model diff --git a/tests/io/test_path.py b/tests/io/test_path.py index 488ff8df06246..832f8ae6639ba 100644 --- a/tests/io/test_path.py +++ b/tests/io/test_path.py @@ -234,11 +234,18 @@ def test_serde_objectstoragepath(self): s = o.serialize() assert s["path"] == path - d = ObjectStoragePath.deserialize(s, 1) - assert o == d + o = ObjectStoragePath(path, my_setting="foo") + s = o.serialize() + assert s["my_setting"] == "foo" + + store = attach("filex", conn_id="mock") + o = ObjectStoragePath(path, store=store) + s = o.serialize() + assert s["store"] == store + def test_serde_store(self): store = attach("file", conn_id="mock") s = store.serialize() From 0b9557742cbe799b53149ea6036459aa91ed2306 Mon Sep 17 00:00:00 2001 From: Bolke de Bruin Date: Sat, 18 Nov 2023 19:51:25 +0100 Subject: [PATCH 12/12] Remove redundant if --- airflow/io/path.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/airflow/io/path.py b/airflow/io/path.py index f308096f67444..9fd51091dc469 100644 --- a/airflow/io/path.py +++ b/airflow/io/path.py @@ -361,9 +361,7 @@ def copy(self, dst: str | ObjectStoragePath, recursive: bool = False, **kwargs) return # remote file -> remote dir - if self.is_file(): - self._cp_file(dst, **kwargs) - return + self._cp_file(dst, **kwargs) def move(self, path: str | ObjectStoragePath, recursive: bool = False, **kwargs) -> None: """Move file(s) from this path to another location.