[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

.github/scripts/cleanup_staging_bucket.py

@@ -1,6 +1,7 @@
 import os
+from datetime import UTC, datetime, timedelta
+
 import boto3
-from datetime import datetime, timedelta, timezone
 
 DELETION_THRESHOLD_DAYS = 7
 # Load config from environment
@@ -25,7 +26,7 @@
 pages = paginator.paginate(Bucket=STAGING_BUCKET)
 
 objects_to_delete = []
-now = datetime.now(timezone.utc)
+now = datetime.now(UTC)
 threshold = now - timedelta(days=DELETION_THRESHOLD_DAYS)
 
 for page in pages:
@@ -50,7 +51,7 @@
     response = client.delete_objects(
         Bucket=STAGING_BUCKET, Delete={"Objects": chunk, "Quiet": True}
     )
-    if "Errors" in response and response["Errors"]:
+    if response.get("Errors"):
         print("  ❌ ERROR during batch deletion:")
         for error in response["Errors"]:
             print(

.github/scripts/publish_script.py

@@ -1,8 +1,8 @@
-import os
 import json
+import os
 import subprocess
-
 from typing import Any
+
 import boto3
 from botocore.exceptions import ClientError
 
@@ -194,7 +194,7 @@ def handle_publications(manifest_data: list[dict[str, Any]]) -> bool:
                 if entry.get("description") == "pending-merge":
                     entry["description"] = commit_details["subject"]
 
-                if "staging_key" in entry and entry["staging_key"]:
+                if entry.get("staging_key"):
                     staging_key = entry.pop("staging_key")
                     final_key = entry["r2_object_key"]
                     print(f"Publishing: {dataset['fileName']} v{entry['version']}")

src/datamanager/__init__.py

@@ -16,5 +16,5 @@
 from . import manifest as manifest
 from .__main__ import app as app  # keeps `python -m datamanager` handy
 
-__all__ = ["app", "core", "manifest", "__version__"]
+__all__ = ["__version__", "app", "core", "manifest"]
 __version__: str = _dist_version("datamanager")

src/datamanager/__main__.py

@@ -1,21 +1,20 @@
 # datamanager/__main__.py
+import re
 import subprocess
-from datetime import datetime, timezone
 import tempfile
-import re
-from dateutil.parser import isoparse
+from collections.abc import Callable
+from datetime import UTC, datetime
 from pathlib import Path
+from typing import Any
 
 import questionary
 import typer
+from dateutil.parser import isoparse
 from rich.console import Console
 from rich.table import Table
 
-from typing import Callable, Optional, Any
-
-from datamanager.config import settings
 from datamanager import core, manifest
-
+from datamanager.config import settings
 
 # Common options for all commands
 COMMON_OPTIONS = dict(
@@ -31,13 +30,13 @@
 def _ask_confirm(ctx: typer.Context, prompt: str, default: bool = False) -> bool:
     if ctx.obj.get("no_prompt"):
         return True
-    result: Optional[bool] = questionary.confirm(prompt, default=default).ask()
+    result: bool | None = questionary.confirm(prompt, default=default).ask()
     return bool(result)  # Cast to bool to avoid NoneType issues
 
 
 def _rel(iso: str) -> str:
     dt = isoparse(iso)
-    delta = datetime.now(timezone.utc) - dt
+    delta = datetime.now(UTC) - dt
     hours = int(delta.total_seconds() // 3600)
     return f"{hours} h ago"
 
@@ -138,7 +137,7 @@ def list_datasets(ctx: typer.Context) -> None:
     console.print(table)
 
 
-def _run_pull_logic(name: str, version: str, output: Optional[Path]) -> None:
+def _run_pull_logic(name: str, version: str, output: Path | None) -> None:
     """The core logic for pulling and verifying a dataset."""
     console.print(f"🔎 Locating version '{version}' for dataset '{name}'...")
     version_entry = manifest.get_version_entry(name, version)
@@ -188,7 +187,7 @@ def pull(
         "-v",
         help="Version to pull (e.g., 'v1'). Defaults to latest.",
     ),
-    output: Optional[Path] = typer.Option(
+    output: Path | None = typer.Option(
         None,
         "--output",
         "-o",
@@ -330,7 +329,7 @@ def _run_prepare_logic(ctx: typer.Context, name: str, file: Path) -> None:
         console.print(f"Change detected! Preparing new version: {new_version}")
 
         console.print("Downloading previous version to generate diff...")
-        diff_git_path: Optional[Path] = None
+        diff_git_path: Path | None = None
         with tempfile.TemporaryDirectory() as tempdir:
             old_path = Path(tempdir) / "prev.sqlite"
             # Download from the PRODUCTION bucket
@@ -355,7 +354,7 @@ def _run_prepare_logic(ctx: typer.Context, name: str, file: Path) -> None:
 
         new_entry = {
             "version": new_version,
-            "timestamp": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
+            "timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
             "sha256": new_hash,
             "r2_object_key": final_r2_key,
             "staging_key": staging_key,
@@ -376,9 +375,7 @@ def _run_prepare_logic(ctx: typer.Context, name: str, file: Path) -> None:
             "history": [
                 {
                     "version": "v1",
-                    "timestamp": datetime.now(timezone.utc)
-                    .isoformat()
-                    .replace("+00:00", "Z"),
+                    "timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
                     "sha256": new_hash,
                     "r2_object_key": f"{Path(Path(name).stem)}/v1-{new_hash}.sqlite",
                     "staging_key": staging_key,
@@ -492,7 +489,7 @@ def _run_rollback_logic(ctx: typer.Context, name: str, to_version: str) -> None:
 
     rollback_entry = {
         "version": new_version,
-        "timestamp": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
+        "timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
         "sha256": target_entry["sha256"],
         "r2_object_key": target_entry["r2_object_key"],
         "diffFromPrevious": None,

src/datamanager/config.py

@@ -1,10 +1,12 @@
 # src/datamanager/config.py
 from __future__ import annotations
+
+import warnings
 from dataclasses import dataclass
 from functools import cached_property
 from pathlib import Path
-from dotenv import find_dotenv, dotenv_values
-import warnings
+
+from dotenv import dotenv_values, find_dotenv
 
 _ENV_PATH = Path(find_dotenv()) if find_dotenv() else None
 _ENV = dotenv_values(_ENV_PATH) if _ENV_PATH else {}

src/datamanager/core.py

@@ -2,26 +2,22 @@
 import difflib
 import hashlib
 import io
+import os
 import shutil
 import sqlite3
 import subprocess
-from pathlib import Path, PurePath
-import os
 import uuid
-
-from botocore.exceptions import ClientError
+from pathlib import Path, PurePath
+from typing import Any, TypedDict
 
 import boto3
-from rich.progress import Progress
+from botocore.exceptions import ClientError
 from rich.console import Console
-
+from rich.progress import Progress
 from types_boto3_s3.client import S3Client
 
-from typing import Any, TypedDict
-
 from datamanager.config import settings
 
-
 console = Console()
 
 

src/datamanager/manifest.py

@@ -6,21 +6,21 @@
 
 import json
 from pathlib import Path
-from typing import Any, Optional
+from typing import Any
 
 from rich.console import Console
 
 from datamanager.config import settings
 
 __all__ = [
-    "read_manifest",
-    "write_manifest",
-    "get_dataset",
     "add_history_entry",
-    "update_latest_history_entry",
-    "get_version_entry",
     "add_new_dataset",
+    "get_dataset",
+    "get_version_entry",
+    "read_manifest",
     "update_dataset",
+    "update_latest_history_entry",
+    "write_manifest",
 ]
 
 # Initialize console for any feedback
@@ -73,7 +73,7 @@ def update_latest_version(name: str, new_version: str) -> None:
     write_manifest(data)
 
 
-def get_dataset(name: str) -> Optional[dict[str, Any]]:
+def get_dataset(name: str) -> dict[str, Any] | None:
     """
     Finds and returns a single dataset from the manifest by its logical name.
 
@@ -158,7 +158,7 @@ def update_latest_history_entry(name: str, final_entry: dict[str, Any]) -> None:
 
 def get_version_entry(
     dataset_name: str, version: str = "latest"
-) -> Optional[dict[str, Any]]:
+) -> dict[str, Any] | None:
     """
     Finds the history entry for a specific version of a dataset.
 

tests/conftest.py

@@ -3,8 +3,9 @@
 import shutil
 import sqlite3
 import subprocess
+from collections.abc import Generator
 from pathlib import Path
-from typing import Any, Generator
+from typing import Any
 
 import pytest
 

tests/test_core.py

@@ -3,10 +3,10 @@
 from pathlib import Path
 from typing import Any
 from unittest.mock import MagicMock
-from pytest_mock import MockerFixture
-import pytest
 
+import pytest
 from botocore.exceptions import ClientError
+from pytest_mock import MockerFixture
 
 from datamanager import core
 from datamanager.config import settings

tests/test_main.py

@@ -5,8 +5,8 @@
 from typer.testing import CliRunner
 
 from datamanager import __main__ as main_app
-from datamanager.__main__ import app
 from datamanager import manifest
+from datamanager.__main__ import app
 from datamanager.config import settings
 
 runner = CliRunner()