Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions dol/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,7 @@ def ihead(store, n=1):
resolve_dir, # to get a full path (resolve ~ and .) and ensure it is a directory
DirReader, # recursive read-only access to directories,
temp_dir, # make a temporary directory,
path_of, # file-system path of a key in a (possibly wrapped) file store
PickleFiles, # CRUD access to pickled files
JsonFiles, # CRUD access to jsob files,
Jsons, # Same as JsonFiles, but with added .json extension handling
Expand Down
61 changes: 47 additions & 14 deletions dol/filesys.py
Original file line number Diff line number Diff line change
Expand Up @@ -1030,24 +1030,57 @@ def _missing_dir_of_key(store, k):
relative path (a relative ``rootdir``) is made relative to the CWD, which is where
the write will put it.
"""
try:
path = path_of(store, k)
except (TypeError, KeyError): # not a file store / key not followable / key rejected
return None
dirname = os.path.dirname(path)
return None if os.path.isdir(dirname) else dirname


def path_of(store, key) -> str:
"""The file-system path of ``key`` in a file-backed ``store``, however it is wrapped.

The key is followed through every layer's ``_id_of_key`` (the same walk as
``dol.inner_most_key``) down to the innermost file-system store, which must accept
the result: a key leaving ``rootdir`` raises ``KeyValidationError`` (a ``KeyError``),
as a read or write of it would. A store with no file-system layer at its core, or
with a layer that maps keys without defining ``_id_of_key`` (so the key cannot be
followed), raises ``TypeError``. Nothing is read or created. The result is an
absolute, normalized path (relative roots resolve against the CWD).

Limit: like ``dol.dig.key_in_scope``, this reads what the layers declare through
``_id_of_key``; a layer that maps keys by other means (``add_path_access``,
``flatten``) is not followed, and a ``filt_iter`` scope is not enforced.

>>> import tempfile
>>> from dol import Files, KeyCodecs
>>> root = tempfile.mkdtemp()
>>> path_of(Files(root), os.path.join('a', 'b.txt')) == os.path.join(root, 'a', 'b.txt')
True
>>> s = KeyCodecs.suffixed('.json')(Files(root)) # wrapped: the outer key is mapped
>>> path_of(s, 'a').endswith('a.json')
True
>>> path_of(Files(root), '../x')
Traceback (most recent call last):
...
dol.filesys.KeyValidationError: 'Key not valid (usually because does not exist or access not permitted): ...'
>>> path_of({}, 'a')
Traceback (most recent call last):
...
TypeError: ...
"""
from dol.dig import inner_most_key

innermost = _innermost_layer(store)
if not isinstance(innermost, FileSysCollection):
return None
_id = inner_most_key(store, k, default=None)
if isinstance(_id, os.PathLike):
_id = os.fspath(_id)
if not isinstance(_id, str):
return None
dirname = os.path.dirname(_id)
if not dirname or os.path.isdir(dirname):
return None
try:
innermost.validate_key(_id)
except KeyValidationError:
return None # the write will raise for this key; make no dirs for it (#107)
return dirname
raise TypeError(
f"{type(innermost).__name__} is not a file-system store: no path for {key!r}"
)
path = key if innermost is store else inner_most_key(store, key)
path = os.fspath(path)
innermost.validate_key(path)
return os.path.abspath(path)


def _innermost_layer(store):
Expand Down
54 changes: 54 additions & 0 deletions dol/tests/test_filesys.py
Original file line number Diff line number Diff line change
Expand Up @@ -563,3 +563,57 @@ def test_bytes_persister_rejects_non_bytes_before_open(tmp_path):
assert s["c"] == b"keep"
s["c"] = bytearray(b"ok")
assert s["c"] == b"ok"


def test_path_of_follows_wrappers_and_validates(tmp_path):
import os

import pytest

from dol import Files, Jsons, KeyCodecs, Pipe, mk_dirs_if_missing, path_of, wrap_kvs
from dol.filesys import FileBytesPersister, KeyValidationError

root = str(tmp_path)
assert path_of(Files(root), os.path.join("a", "b.txt")) == os.path.join(
root, "a", "b.txt"
)
suffixed = KeyCodecs.suffixed(".json")(Files(root))
assert path_of(suffixed, "a") == os.path.join(root, "a.json")
upper = wrap_kvs(Files(root), id_of_key=str.lower, key_of_id=str.upper)
assert path_of(upper, "ABC") == os.path.join(root, "abc")
stacked = Pipe(KeyCodecs.suffixed(".x"), mk_dirs_if_missing)(Files(root))
assert path_of(stacked, "k") == os.path.join(root, "k.x")
assert path_of(Jsons(root), "k") == os.path.join(root, "k.json")
# a bare leaf: keys already are paths
leaf = FileBytesPersister(root)
assert path_of(leaf, os.path.join(root, "z")) == os.path.join(root, "z")
with pytest.raises(KeyError):
path_of(Files(root), os.path.join("..", "x"))
with pytest.raises(KeyValidationError):
path_of(suffixed, os.path.join("..", "x"))
with pytest.raises(KeyValidationError):
path_of(leaf, os.path.join(root, "..", "x"))
with pytest.raises(TypeError):
path_of({}, "a")
assert os.listdir(root) == [] # nothing created


def test_path_of_relative_root_is_absolute(tmp_path, monkeypatch):
import os

from dol import Files, path_of

monkeypatch.chdir(tmp_path)
got = path_of(Files("sub"), "a")
assert os.path.isabs(got) and got == os.path.abspath(os.path.join("sub", "a"))


def test_path_of_allow_escape(tmp_path):
import os

from dol import Files, path_of

root = tmp_path / "r"
root.mkdir()
s = Files(str(root), allow_escape=True)
assert path_of(s, os.path.join("..", "x")) == str(tmp_path / "x")
Loading