From 625905ed15596148379a7b2f20b9a976f78008bd Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Wed, 12 Aug 2026 16:07:05 +0000 Subject: [PATCH] Let a writer keep a half-written directory out of an upload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `exclude` matches glob patterns against a path, so it cannot say "a directory that contains `.unfinished`" — that is a fact about the directory's contents, and the path carries no sign of it. A writer marking its in-progress output therefore had no way to keep it out of the destination, and the workarounds are a move or a post-hoc cleanup, which lose the artifact or the tidiness they were meant to protect. `skip_dirs_containing` names those marker files. The local scan skips a directory holding one, with its whole subtree, on every upload the registration performs — so a directory that finishes and drops its marker uploads on the next pass, and one that never finishes never leaves the machine. Upload half only, deliberately: a directory the destination already holds is not deleted there for gaining a marker. Ticket: Positronic-Robotics/internal#388 #refs --- CHANGELOG.md | 11 ++++++ pos3/__init__.py | 51 +++++++++++++++++++++++---- pos3/cli.py | 4 +-- pos3/profiles.py | 6 ++-- pyproject.toml | 2 +- tests/test_cli.py | 32 +++++------------ tests/test_s3.py | 90 ++++++++++++++++++++++++++++------------------- uv.lock | 2 +- 8 files changed, 122 insertions(+), 76 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0eaecfe..84eddd0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,16 @@ # Changelog +## [0.4.0] - 2026-08-12 + +### Added +- `skip_dirs_containing` on `upload()` and `sync()`: filenames that mark a local directory as + not-to-be-uploaded. A directory holding one is skipped with its whole subtree, so a writer keeps + its half-written output out of the destination by leaving a marker in it. + - Expresses what `exclude` cannot: `exclude` matches glob patterns against a path, and "a + directory that contains `.unfinished`" is a fact about the directory's CONTENTS. + - Upload half only. A directory the destination already holds is not deleted there for gaining a + marker; one that loses its marker uploads on the next pass. + ## [0.3.1] - 2026-05-21 ### Added diff --git a/pos3/__init__.py b/pos3/__init__.py index 121256b..53d89f5 100644 --- a/pos3/__init__.py +++ b/pos3/__init__.py @@ -141,7 +141,14 @@ class FileInfo: is_dir: bool # True if this represents a directory -def _scan_local(path: Path) -> Iterator[FileInfo]: +def _scan_local(path: Path, skip_dirs_containing: list[str] | None = None) -> Iterator[FileInfo]: + """Walk `path`, yielding every directory before its children. + + `skip_dirs_containing` names files that mark a directory as not-to-be-copied: a directory + holding one is skipped with its whole subtree, and the writer that put the marker there is + what decides. A glob cannot express this — the mark is in the directory's CONTENTS, not in + its path — which is why it is a scan-time rule rather than an `exclude` pattern. + """ if not path.exists(): return @@ -155,6 +162,8 @@ def _scan_local(path: Path) -> Iterator[FileInfo]: relative = p.relative_to(base).as_posix() if p != base else "" if p.is_dir(): + if skip_dirs_containing and any((p / marker).exists() for marker in skip_dirs_containing): + continue # Always yield directories, including the root (relative_path='') yield FileInfo(relative_path=relative, size=0, is_dir=True) stack.extend(p.iterdir()) @@ -261,6 +270,7 @@ class _UploadRegistration: sync_on_error: bool exclude: list[str] | None profile: Profile | None = None + skip_dirs_containing: list[str] | None = None last_sync: float = 0.0 # The user's original URL, preserving trailing-slash intent. _sync_uploads # parses it raw so `s3://bucket/data/` skips head_object('data') in @@ -279,6 +289,7 @@ def __eq__(self, other): and self.sync_on_error == other.sync_on_error and self.exclude == other.exclude and self.profile == other.profile + and self.skip_dirs_containing == other.skip_dirs_containing ) @@ -431,6 +442,7 @@ def upload( sync_on_error, exclude: list[str] | None = None, profile: str | Profile | None = None, + skip_dirs_containing: list[str] | None = None, ) -> Path: """ Register (and perform if needed) an upload from a local directory or file to a remote S3 bucket path. @@ -443,6 +455,8 @@ def upload( sync_on_error (bool): If True, attempts to sync files even when encountering errors. exclude (list[str] | None): List of glob patterns to exclude from upload. profile: S3 profile name or Profile config for custom endpoints. + skip_dirs_containing (list[str] | None): Filenames that mark a local directory as + not-to-be-uploaded; one holding any of them is skipped with its subtree. Returns: Path: The canonical local path associated with this upload registration. @@ -474,6 +488,7 @@ def upload( sync_on_error=sync_on_error, exclude=exclude, profile=effective_profile, + skip_dirs_containing=skip_dirs_containing, last_sync=0, raw_remote=remote, ) @@ -606,6 +621,7 @@ def sync( sync_on_error: bool, exclude: list[str] | None = None, profile: str | Profile | None = None, + skip_dirs_containing: list[str] | None = None, ) -> Path: # Let download() and upload() handle profile resolution and normalization local_path = self.download(remote, local, delete_local, exclude, profile) @@ -616,7 +632,9 @@ def sync( effective_profile = self._effective_profile(profile, remote) # Unregister the download to allow upload registration for the same remote self._downloads.pop((normalized, effective_profile), None) - return self.upload(remote, local_path, interval, delete_remote, sync_on_error, exclude, profile) + return self.upload( + remote, local_path, interval, delete_remote, sync_on_error, exclude, profile, skip_dirs_containing + ) def ls(self, prefix: str, recursive: bool = False, profile: str | Profile | None = None) -> list[str]: """Lists objects under the given prefix, working for both local directories and S3 prefixes.""" @@ -743,7 +761,7 @@ def _final_sync(self, had_error: bool = False) -> None: self._sync_uploads(uploads) def _sync_uploads(self, registrations: Iterable[_UploadRegistration]) -> None: - tasks: list[tuple[str, Path, bool, list[str] | None, Profile | None]] = [] + tasks: list[tuple[str, Path, bool, list[str] | None, Profile | None, list[str] | None]] = [] for registration in registrations: if registration.local_path.exists(): tasks.append( @@ -757,6 +775,7 @@ def _sync_uploads(self, registrations: Iterable[_UploadRegistration]) -> None: registration.delete, registration.exclude, registration.profile, + registration.skip_dirs_containing, ) ) @@ -767,12 +786,12 @@ def _sync_uploads(self, registrations: Iterable[_UploadRegistration]) -> None: to_remove: list[tuple[str, str, Profile | None]] = [] total_bytes = 0 - for remote, local_path, delete, exclude, profile in tasks: + for remote, local_path, delete, exclude, profile, skip_dirs_containing in tasks: logger.debug("Syncing upload: %s from %s (delete=%s)", remote, local_path, delete) scan_bucket, scan_prefix = _parse_s3_url(remote) bucket, out_prefix = _parse_s3_url(_normalize_s3_url(remote)) to_copy, to_delete = _compute_sync_diff( - _filter_fileinfo(_scan_local(local_path), exclude), + _filter_fileinfo(_scan_local(local_path, skip_dirs_containing), exclude), _filter_fileinfo(self._scan_s3(scan_bucket, scan_prefix, profile), exclude), ) @@ -1118,6 +1137,7 @@ def upload( sync_on_error: bool = False, exclude: list[str] | None = None, profile: str | Profile | None = None, + skip_dirs_containing: list[str] | None = None, ) -> Path: """ Register a local path for upload. Uploads on exit and optionally in background. @@ -1129,12 +1149,15 @@ def upload( delete: If True (default), deletes S3 files NOT present locally. sync_on_error: If True, syncs even if the context exits with an exception. profile: S3 profile name or Profile config for custom endpoints. + skip_dirs_containing: Filenames marking a local directory as not-to-be-uploaded; a + directory holding any of them is skipped with its subtree, so a writer can keep its + half-written output out of the destination by leaving a marker in it. Returns: Path to the local directory/file. """ mirror_obj = _require_active_mirror() - return mirror_obj.upload(remote, local, interval, delete, sync_on_error, exclude, profile) + return mirror_obj.upload(remote, local, interval, delete, sync_on_error, exclude, profile, skip_dirs_containing) def sync( @@ -1146,6 +1169,7 @@ def sync( sync_on_error: bool = False, exclude: list[str] | None = None, profile: str | Profile | None = None, + skip_dirs_containing: list[str] | None = None, ) -> Path: """ Bi-directional helper. Performs download() then registers upload(). @@ -1154,12 +1178,25 @@ def sync( delete_local: Cleanup local files during download. delete_remote: Cleanup remote files during upload. profile: S3 profile name or Profile config for custom endpoints. + skip_dirs_containing: Filenames marking a local directory as not-to-be-uploaded. Applies to + the upload half only — a directory the destination already holds is not deleted there + for gaining a marker, and one that loses its marker uploads on the next pass. Returns: Path to the local directory/file. """ mirror_obj = _require_active_mirror() - return mirror_obj.sync(remote, local, interval, delete_local, delete_remote, sync_on_error, exclude, profile) + return mirror_obj.sync( + remote, + local, + interval, + delete_local, + delete_remote, + sync_on_error, + exclude, + profile, + skip_dirs_containing, + ) def ls(prefix: str, recursive: bool = False, profile: str | Profile | None = None) -> list[str]: diff --git a/pos3/cli.py b/pos3/cli.py index 7378496..d7a0b89 100644 --- a/pos3/cli.py +++ b/pos3/cli.py @@ -209,9 +209,7 @@ def main(argv: list[str] | None = None) -> int: # because they can't see the flags the subcommand actually exposes. namespace, leftover = parser.parse_known_args(argv) if leftover: - subparsers_action = next( - a for a in parser._actions if isinstance(a, argparse._SubParsersAction) - ) + subparsers_action = next(a for a in parser._actions if isinstance(a, argparse._SubParsersAction)) sub = subparsers_action.choices.get(namespace.command, parser) sub.error(f"unrecognized arguments: {' '.join(leftover)}") args = namespace diff --git a/pos3/profiles.py b/pos3/profiles.py index beb7f27..f4958cc 100644 --- a/pos3/profiles.py +++ b/pos3/profiles.py @@ -189,8 +189,7 @@ def _load_profile_registry(path: Path | None = None, force: bool = False) -> Non # Build all profiles first so a malformed entry doesn't leave the # registry half-loaded with a sticky _REGISTRY_LOADED flag. new_profiles = { - name: _profile_from_config(name, cfg, registry_path) - for name, cfg in data.get("profiles", {}).items() + name: _profile_from_config(name, cfg, registry_path) for name, cfg in data.get("profiles", {}).items() } else: new_profiles = {} @@ -245,8 +244,7 @@ def _url_profile(s3_url: str) -> str | None: return None if not parsed.username: raise ValueError( - f"Empty profile selector in S3 URL: {s3_url!r}. " - "Use s3://@bucket/key or omit the '@'." + f"Empty profile selector in S3 URL: {s3_url!r}. Use s3://@bucket/key or omit the '@'." ) return parsed.username diff --git a/pyproject.toml b/pyproject.toml index 4d6627e..abfd7fe 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "pos3" -version = "0.3.1" +version = "0.4.0" description = "S3 Simple Sync - Make using S3 as simple as using local files" readme = "README.md" requires-python = ">=3.11" diff --git a/tests/test_cli.py b/tests/test_cli.py index bcdbd34..eb43b85 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -427,18 +427,14 @@ def test_download_dry_run_with_delete_emits_delete_lines(self, mock_boto_client, orphan = local_dir / "orphan.txt" orphan.write_text("orphan") - rc = main( - ["download", "-n", "s3://bucket/data", "--local", str(local_dir), "--delete"] - ) + rc = main(["download", "-n", "s3://bucket/data", "--local", str(local_dir), "--delete"]) assert rc == 0 # Dry-run must not touch the filesystem. assert orphan.exists() captured = capsys.readouterr() - delete_lines = [ - line for line in captured.out.splitlines() if line.startswith("delete:") - ] + delete_lines = [line for line in captured.out.splitlines() if line.startswith("delete:")] assert any(str(orphan) in line for line in delete_lines) mock_s3.download_file.assert_not_called() @@ -456,9 +452,7 @@ def test_upload_dry_run_prints_plan_and_does_not_transfer(self, mock_boto_client captured = capsys.readouterr() assert rc == 0 mock_s3.upload_file.assert_not_called() - upload_lines = [ - line for line in captured.out.splitlines() if line.startswith("upload:") - ] + upload_lines = [line for line in captured.out.splitlines() if line.startswith("upload:")] assert len(upload_lines) == 1 assert "s3://bucket/data/file.txt" in upload_lines[0] assert str(src / "file.txt") in upload_lines[0] @@ -473,17 +467,13 @@ def test_upload_dry_run_with_delete_emits_remote_delete_lines(self, mock_boto_cl src.mkdir() (src / "file.txt").write_text("content") - rc = main( - ["upload", "-n", "s3://bucket/data", "--local", str(src), "--delete"] - ) + rc = main(["upload", "-n", "s3://bucket/data", "--local", str(src), "--delete"]) captured = capsys.readouterr() assert rc == 0 mock_s3.upload_file.assert_not_called() mock_s3.delete_object.assert_not_called() - delete_lines = [ - line for line in captured.out.splitlines() if line.startswith("delete:") - ] + delete_lines = [line for line in captured.out.splitlines() if line.startswith("delete:")] assert any("s3://bucket/data/remote_only.txt" in line for line in delete_lines) def test_dry_run_not_accepted_on_ls(self): @@ -522,9 +512,7 @@ def test_ls_handles_client_error(self, mock_boto_client, capsys): Python traceback.""" mock_s3 = Mock() mock_boto_client.return_value = mock_s3 - mock_s3.head_object.side_effect = ClientError( - {"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject" - ) + mock_s3.head_object.side_effect = ClientError({"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject") rc = main(["ls", "s3://bucket/key"]) @@ -541,9 +529,7 @@ def test_download_handles_client_error_during_scan(self, mock_boto_client, capsy traceback.""" mock_s3 = Mock() mock_boto_client.return_value = mock_s3 - mock_s3.head_object.side_effect = ClientError( - {"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject" - ) + mock_s3.head_object.side_effect = ClientError({"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject") rc = main(["download", "s3://bucket/data", "--local", str(tmp_path / "dst")]) @@ -558,9 +544,7 @@ def test_dry_run_handles_client_error_during_plan(self, mock_boto_client, capsys _scan_s3 the same way.""" mock_s3 = Mock() mock_boto_client.return_value = mock_s3 - mock_s3.head_object.side_effect = ClientError( - {"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject" - ) + mock_s3.head_object.side_effect = ClientError({"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject") rc = main(["download", "-n", "s3://bucket/data", "--local", str(tmp_path / "dst")]) diff --git a/tests/test_s3.py b/tests/test_s3.py index 10736a8..a798b11 100644 --- a/tests/test_s3.py +++ b/tests/test_s3.py @@ -224,9 +224,7 @@ def test_upload_delete_directory_marker_trailing_slash(self, mock_boto_client): # The directory marker should be deleted with trailing slash delete_calls = mock_s3.delete_object.call_args_list deleted_keys = [call[1]["Key"] for call in delete_calls] - assert "output/subdir/" in deleted_keys, ( - f"Expected delete of 'output/subdir/' but got: {deleted_keys}" - ) + assert "output/subdir/" in deleted_keys, f"Expected delete of 'output/subdir/' but got: {deleted_keys}" @patch(BOTO3_PATCH_TARGET) def test_background_sync_uploads_repeatedly(self, mock_boto_client): @@ -767,9 +765,7 @@ def test_plan_download_lists_files_to_copy(self, mock_boto_client): with s3.mirror(cache_root=tmpdir, show_progress=False) as _: from pos3 import _require_active_mirror - plan = _require_active_mirror().plan_download( - "s3://bucket/data", local=str(local_dir) - ) + plan = _require_active_mirror().plan_download("s3://bucket/data", local=str(local_dir)) sources = [src for src, _ in plan.to_copy] assert "s3://bucket/data/file.txt" in sources @@ -794,9 +790,7 @@ def test_plan_download_lists_orphans_in_to_delete(self, mock_boto_client): with s3.mirror(cache_root=tmpdir, show_progress=False): from pos3 import _require_active_mirror - plan = _require_active_mirror().plan_download( - "s3://bucket/data", local=str(local_dir) - ) + plan = _require_active_mirror().plan_download("s3://bucket/data", local=str(local_dir)) # Dry-plan is read-only. assert orphan.exists() @@ -815,9 +809,7 @@ def test_plan_upload_lists_files_to_copy(self, mock_boto_client): with s3.mirror(cache_root=tmpdir, show_progress=False): from pos3 import _require_active_mirror - plan = _require_active_mirror().plan_upload( - "s3://bucket/data", local=str(src) - ) + plan = _require_active_mirror().plan_upload("s3://bucket/data", local=str(src)) destinations = [dst for _, dst in plan.to_copy] assert destinations == ["s3://bucket/data/file.txt"] @@ -851,9 +843,7 @@ def test_plan_download_normalizes_trailing_slash_in_url(self, mock_boto_client): with s3.mirror(cache_root=tmpdir, show_progress=False): from pos3 import _require_active_mirror - plan = _require_active_mirror().plan_download( - "s3://bucket/data/", local=str(local_dir) - ) + plan = _require_active_mirror().plan_download("s3://bucket/data/", local=str(local_dir)) sources = [src for src, _ in plan.to_copy] assert sources == ["s3://bucket/data/file.txt"] @@ -878,9 +868,7 @@ def test_plan_download_trailing_slash_forces_directory_listing(self, mock_boto_c with s3.mirror(cache_root=tmpdir, show_progress=False): from pos3 import _require_active_mirror - plan = _require_active_mirror().plan_download( - "s3://bucket/data/", local=str(Path(tmpdir) / "dst") - ) + plan = _require_active_mirror().plan_download("s3://bucket/data/", local=str(Path(tmpdir) / "dst")) sources = [src for src, _ in plan.to_copy] assert sources == ["s3://bucket/data/file.txt"] @@ -898,9 +886,7 @@ def test_plan_upload_normalizes_trailing_slash_in_url(self, mock_boto_client): with s3.mirror(cache_root=tmpdir, show_progress=False): from pos3 import _require_active_mirror - plan = _require_active_mirror().plan_upload( - "s3://bucket/data/", local=str(src) - ) + plan = _require_active_mirror().plan_upload("s3://bucket/data/", local=str(src)) destinations = [dst for _, dst in plan.to_copy] assert destinations == ["s3://bucket/data/file.txt"] @@ -922,9 +908,7 @@ def test_download_trailing_slash_transfers_directory_contents(self, mock_boto_cl mock_s3.head_object.return_value = {"ContentLength": 100} mock_paginator = Mock() mock_s3.get_paginator.return_value = mock_paginator - mock_paginator.paginate.return_value = [ - {"Contents": [{"Key": "data/file.txt", "Size": 5}]} - ] + mock_paginator.paginate.return_value = [{"Contents": [{"Key": "data/file.txt", "Size": 5}]}] with tempfile.TemporaryDirectory() as tmpdir: local = Path(tmpdir) / "dst" @@ -945,9 +929,7 @@ def test_upload_trailing_slash_scans_directory_for_delete(self, mock_boto_client mock_s3.head_object.return_value = {"ContentLength": 100} mock_paginator = Mock() mock_s3.get_paginator.return_value = mock_paginator - mock_paginator.paginate.return_value = [ - {"Contents": [{"Key": "data/orphan.txt", "Size": 5}]} - ] + mock_paginator.paginate.return_value = [{"Contents": [{"Key": "data/orphan.txt", "Size": 5}]}] with tempfile.TemporaryDirectory() as tmpdir: source = Path(tmpdir) / "src" @@ -983,9 +965,7 @@ def test_plan_upload_empty_when_source_missing(self, mock_boto_client): with s3.mirror(cache_root=tmpdir, show_progress=False): from pos3 import _require_active_mirror - plan = _require_active_mirror().plan_upload( - "s3://bucket/data", local=str(missing) - ) + plan = _require_active_mirror().plan_upload("s3://bucket/data", local=str(missing)) assert plan.to_copy == [] assert plan.to_delete == [] @@ -1070,9 +1050,7 @@ def test_app_exception_survives_scan_client_error_in_cleanup(self, mock_boto_cli # _scan_s3 → _list_s3_objects calls head_object first. A 403 here # propagates through the scan iterator, not through a worker future, # so the previous TransferError-only catch would have unmasked it. - mock_s3.head_object.side_effect = ClientError( - {"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject" - ) + mock_s3.head_object.side_effect = ClientError({"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject") class AppError(Exception): pass @@ -1368,9 +1346,9 @@ def test_prefix_boundary_prevents_spurious_matches(self, mock_boto_client): paginator_calls = mock_s3.get_paginator.return_value.paginate.call_args_list assert len(paginator_calls) == 1 call_kwargs = paginator_calls[0][1] - assert ( - call_kwargs["Prefix"] == "data/" - ), f"Expected Prefix='data/' but got Prefix='{call_kwargs['Prefix']}'" + assert call_kwargs["Prefix"] == "data/", ( + f"Expected Prefix='data/' but got Prefix='{call_kwargs['Prefix']}'" + ) @patch(BOTO3_PATCH_TARGET) def test_prefix_boundary_with_trailing_slash(self, mock_boto_client): @@ -1922,3 +1900,43 @@ def test_url_unknown_profile_hard_error(self, mock_boto_client): with s3.mirror(cache_root=tmpdir, show_progress=False): with pytest.raises(ValueError, match="Unknown profile"): s3.download("s3://ghost@bucket/data") + + +class TestSkipDirsContaining: + """A directory is kept out of an upload by what is INSIDE it, which no glob can say.""" + + def test_scan_skips_a_marked_directory_and_its_subtree(self, tmp_path): + from pos3 import _scan_local + + (tmp_path / "done" / "inner").mkdir(parents=True) + (tmp_path / "done" / "inner" / "a.bin").write_text("a") + (tmp_path / "open").mkdir() + (tmp_path / "open" / ".unfinished").write_text("") + (tmp_path / "open" / "b.bin").write_text("b") + + seen = {i.relative_path for i in _scan_local(tmp_path, [".unfinished"])} + + assert "done/inner/a.bin" in seen + assert not any(p.startswith("open") for p in seen) + + def test_scan_keeps_everything_when_no_marker_is_named(self, tmp_path): + from pos3 import _scan_local + + (tmp_path / "open").mkdir() + (tmp_path / "open" / ".unfinished").write_text("") + + seen = {i.relative_path for i in _scan_local(tmp_path)} + + assert "open" in seen and "open/.unfinished" in seen + + def test_a_directory_that_loses_its_marker_is_scanned_again(self, tmp_path): + from pos3 import _scan_local + + (tmp_path / "ep").mkdir() + marker = tmp_path / "ep" / ".unfinished" + marker.write_text("") + (tmp_path / "ep" / "a.bin").write_text("a") + + assert not any(i.relative_path.startswith("ep") for i in _scan_local(tmp_path, [".unfinished"])) + marker.unlink() + assert "ep/a.bin" in {i.relative_path for i in _scan_local(tmp_path, [".unfinished"])} diff --git a/uv.lock b/uv.lock index 9a2d586..907a1e0 100644 --- a/uv.lock +++ b/uv.lock @@ -223,7 +223,7 @@ wheels = [ [[package]] name = "pos3" -version = "0.3.0" +version = "0.3.1" source = { editable = "." } dependencies = [ { name = "boto3" },