Skip to content

Commit

Permalink
fix(stat): return BlobStat for all pathy paths
Browse files Browse the repository at this point in the history
- implement `stat` in `BasePath` and yield `BlobStat`s instead of `os.stat_result`.
- implement all the pathlib file mode checking helpers because they use `self.stat()` internally which doesn't work with BlobStats.
- add tests for base path helpers
- update CLI test to assert that stats are BlobStat now

BREAKING CHANGE: Previously when using Pathy.fluid paths that point to local file system paths, Pathy would return an `os.stat_result` rather than a `BlobStat`. This made it difficulty to treat mixed paths consistently.

Now Pathy returns a BlobStat structure for local and remote paths.

If you need to use `os.stat_result` you can still call `os.stat(my_path)` to access it.
  • Loading branch information
justindujardin committed Nov 23, 2022
1 parent e310e5e commit 89e7a35
Show file tree
Hide file tree
Showing 3 changed files with 129 additions and 5 deletions.
69 changes: 69 additions & 0 deletions pathy/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,10 +6,12 @@
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from errno import EBADF, ELOOP, ENOENT, ENOTDIR
from io import DEFAULT_BUFFER_SIZE
from pathlib import _PosixFlavour # type:ignore
from pathlib import _WindowsFlavour # type:ignore
from pathlib import Path, PurePath
from stat import S_ISBLK, S_ISCHR, S_ISFIFO, S_ISSOCK
from typing import (
IO,
Any,
Expand Down Expand Up @@ -365,6 +367,73 @@ def iterdir(self: Any) -> Generator["BasePath", None, None]:
for blob in blobs:
yield self / blob.name

def stat(self: "BasePath") -> BlobStat: # type:ignore[override]
"""Iterate over the blobs found in the given bucket or blob prefix path."""
stat = super().stat()
return BlobStat(
name=self.name, size=stat.st_size, last_modified=int(stat.st_mtime)
)

# Stat helpers

def _check_mode(self: "BasePath", mode_fn: Callable[[int], bool]) -> bool:
"""
Check the mode against a stat.S_IS[MODE] function.
This ignores OS-specific errors that are raised when a path does
not exist, or has some invalid attribute (e.g. a bad symlink).
"""
try:
return mode_fn(os.stat(self).st_mode)
except OSError as exception:
# Ignorable error codes come from pathlib.py
#
error = getattr(exception, "errno", None)
errors = (ENOENT, ENOTDIR, EBADF, ELOOP)
win_error = getattr(exception, "winerror", None)
win_errors = (
21, # ERROR_NOT_READY - drive exists but is not accessible
123, # ERROR_INVALID_NAME - fix for bpo-35306
1921, # ERROR_CANT_RESOLVE_FILENAME - broken symlink points to self
)
if error not in errors and win_error not in win_errors:
raise
return False
except ValueError:
return False

def is_dir(self: "BasePath") -> bool:
"""Whether this path is a directory."""
return os.path.isdir(self)

def is_file(self: "BasePath") -> bool:
"""Whether this path is a file."""
return os.path.isfile(self)

def is_mount(self: "BasePath") -> bool:
"""Check if this path is a POSIX mount point"""
return os.path.ismount(self)

def is_symlink(self: "BasePath") -> bool:
"""Whether this path is a symbolic link."""
return os.path.islink(self)

def is_block_device(self: "BasePath") -> bool:
"""Whether this path is a block device."""
return self._check_mode(S_ISBLK)

def is_char_device(self: "BasePath") -> bool:
"""Whether this path is a character device."""
return self._check_mode(S_ISCHR)

def is_fifo(self: "BasePath") -> bool:
"""Whether this path is a FIFO."""
return self._check_mode(S_ISFIFO)

def is_socket(self: "BasePath") -> bool:
"""Whether this path is a socket."""
return self._check_mode(S_ISSOCK)


class BucketsAccessor:
"""Path access for python < 3.11"""
Expand Down
52 changes: 52 additions & 0 deletions pathy/_tests/test_base.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
import tempfile
from errno import ENOTDIR
from pathlib import Path
from typing import Any

Expand Down Expand Up @@ -102,6 +104,56 @@ def test_base_symlink_to() -> None:
path.symlink_to("file_name")


def test_base_path_check_mode() -> None:
tmp_dir = tempfile.mkdtemp()
root = Pathy.fluid(tmp_dir)

assert root.is_dir() is True

# Ignores OSErrors about not a directory and returns false
def not_dir_fn(mode: int) -> bool:
err = OSError()
err.errno = ENOTDIR
raise err

assert root._check_mode(not_dir_fn) is False

# Ignores ValueError
def value_error_fn(mode: int) -> bool:
raise ValueError("oops")

assert root._check_mode(value_error_fn) is False

# Raises other unrelated exceptions
def other_error_fn(mode: int) -> bool:
raise BaseException()

with pytest.raises(BaseException):
root._check_mode(other_error_fn)


def test_base_path_stat_helpers() -> None:
tmp_dir = tempfile.mkdtemp()
root = Pathy.fluid(tmp_dir)

assert root.is_dir() is True

file = root / "file.txt"
file.write_text("hello world")

assert file.is_file() is True
assert file.is_dir() is False
assert file.is_mount() is False
assert file.is_symlink() is False
assert file.is_block_device() is False
assert file.is_char_device() is False
assert file.is_fifo() is False
assert file.is_socket() is False

file.unlink()
root.rmdir()


def test_pathy_mro() -> None:
assert PurePathy in Pathy.mro()
assert Path in Pathy.mro()
Expand Down
13 changes: 8 additions & 5 deletions pathy/_tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import pytest
from typer.testing import CliRunner

from pathy import Pathy
from pathy import BlobStat, Pathy
from pathy.cli import app

from .conftest import ENV_ID, TEST_ADAPTERS
Expand Down Expand Up @@ -279,14 +279,17 @@ def test_cli_ls_diff_years_modified() -> None:
old_path = root / "old_file.txt"
old_path.write_text("old")
old_stat = old_path.stat()
assert isinstance(old_stat, os.stat_result), "expect local file"
assert isinstance(old_stat, BlobStat)
one_year = 31556926 # seconds
assert old_stat.last_modified is not None
os.utime(
str(old_path), (old_stat.st_atime - one_year, old_stat.st_mtime - one_year)
str(old_path),
(old_stat.last_modified - one_year, old_stat.last_modified - one_year),
)
new_old_stat = old_path.stat()
assert isinstance(new_old_stat, os.stat_result), "expect local file"
assert int(old_stat.st_mtime) == int(new_old_stat.st_mtime + one_year)
assert new_old_stat.last_modified is not None
assert isinstance(new_old_stat, BlobStat)
assert int(old_stat.last_modified) == int(new_old_stat.last_modified + one_year)

result = runner.invoke(app, ["ls", "-l", str(root)])
assert result.exit_code == 0
Expand Down

0 comments on commit 89e7a35

Please sign in to comment.