diff --git a/.env.example b/.env.example index b6dcffd..016408f 100644 --- a/.env.example +++ b/.env.example @@ -13,3 +13,7 @@ MINIO_CREATE_BUCKET_IF_MISSING=true POSTGRES_URI=postgresql://localhost/mydb POSTGRES_TABLE=users POSTGRES_PRIMARY_KEY=id + +# File-backed queue (no optional extra) +FILE_QUEUE_PATH=/tmp/python-repositories-queue.jsonl +# FILE_QUEUE_MAX_AGE_HOURS=24 diff --git a/README.md b/README.md index 0755aa6..d791918 100644 --- a/README.md +++ b/README.md @@ -6,11 +6,11 @@ Subclass an adapter in your own repository to add domain-specific methods while ## Architecture -| Layer | Responsibility | -| ---------------- | ------------------------------------------------------------------------------------ | -| **Interfaces** | Abstract contracts for connection, context, and CRUD | -| **Adapters** | Technology-specific base classes (`RedisAdapter`, `MinioAdapter`, `PostgresAdapter`) | -| **Your project** | Subclass an adapter and add domain methods | +| Layer | Responsibility | +| ---------------- | ------------------------------------------------------------------------------------------------------- | +| **Interfaces** | Abstract contracts for connection, context, CRUD, and queues | +| **Adapters** | Technology-specific base classes (`RedisAdapter`, `MinioAdapter`, `PostgresAdapter`, queue adapters) | +| **Your project** | Subclass an adapter and add domain methods | Each public interface is a `@runtime_checkable` `Protocol` with `@abstractmethod` members. **Subclass an adapter** when you need connection management and shared behavior — explicit subclasses get runtime instantiation guards and inherited default methods (e.g. `scan_keys`). **Type-annotate against an interface** when you want loose coupling — any object with the right methods satisfies the contract for mypy and `isinstance()` checks, without inheriting from this package. @@ -22,7 +22,7 @@ The current API is synchronous. Async repository interfaces and adapters may be ## Optional dependencies -Repository **interfaces** import with the base package. **Adapters** require the matching extra; importing an adapter without its extra raises `ImportError` with install instructions. +Repository **interfaces** import with the base package. Networked **adapters** (`RedisAdapter`, `MinioAdapter`, `PostgresAdapter`) require the matching extra; importing one without its extra raises `ImportError` with install instructions. Queue adapters (`MemoryQueueAdapter`, `FileBackedQueueAdapter`) need no extra. Install with the extras you need: @@ -76,14 +76,50 @@ Requires PostgreSQL 10 or later; tested against PostgreSQL 16 in CI. Tables are | `POSTGRES_TABLE` | Table name for CRUD operations | | `POSTGRES_PRIMARY_KEY` | Primary key column name (default: `id`) | -Copy [`.env.example`](.env.example) to `.env` for local development. `RedisConfig.from_env()`, `MinioConfig.from_env()`, and `PostgresConfig.from_env()` load `.env` automatically when resolving configuration from the environment. +### File-backed queue (`QueueRepositoryInterface`) + +In-process and disk-backed FIFO queues for buffering dict items. No optional extra required. + +| Adapter | Role | +| ------------------------ | -------------------------------------------------------------------- | +| `MemoryQueueAdapter` | Thread-safe in-memory buffer with optional dedup and age eviction | +| `FileBackedQueueAdapter` | Mirrors memory to a JSONL file; replays on `connect()`, compacts on dequeue/evict | + +| Environment variable | Description | +| ---------------------------- | -------------------------------------------------------- | +| `FILE_QUEUE_PATH` | Path to the JSONL queue file | +| `FILE_QUEUE_MAX_AGE_HOURS` | Retention window in hours (default: `24`) | + +Dedup keys and the age field name are domain-specific: pass them when constructing `FileQueueConfig` (or as kwargs to `FileQueueConfig.from_env(...)`). Callers choose the file path and item schema. + +```python +from pathlib import Path + +from python_repositories import FileBackedQueueAdapter, FileQueueConfig + +config = FileQueueConfig( + path=Path("/var/lib/my-service/queue/items.jsonl"), + max_age_hours=24, + dedup_keys=("id",), + age_key="created_at", +) +with FileBackedQueueAdapter(config=config) as queue: + queue.enqueue([{"id": 1, "created_at": "2024-01-01T00:00:00+00:00"}]) + batch = queue.dequeue_batch(max_items=100) +``` + +Copy [`.env.example`](.env.example) to `.env` for local development. `RedisConfig.from_env()`, `MinioConfig.from_env()`, `PostgresConfig.from_env()`, and `FileQueueConfig.from_env()` load `.env` automatically when resolving configuration from the environment. ## Configuration injection -Adapters accept optional `config` and `client` keyword arguments for explicit setup and testing: +Adapters accept optional `config` and `client` keyword arguments for explicit setup and testing. Queue adapters accept `config` (and an optional injected `memory` buffer for `FileBackedQueueAdapter`): ```python +from pathlib import Path + from python_repositories import ( + FileBackedQueueAdapter, + FileQueueConfig, MinioAdapter, MinioConfig, PostgresAdapter, @@ -110,6 +146,13 @@ postgres = PostgresAdapter( primary_key="user_id", ) ) +queue = FileBackedQueueAdapter( + config=FileQueueConfig( + path=Path("/var/lib/my-service/queue/items.jsonl"), + dedup_keys=("id",), + age_key="created_at", + ) +) ``` When both `config` and `client` are provided, `connect()` skips client creation (the caller owns the client lifecycle). `config` is required whenever `client` is injected. @@ -186,12 +229,16 @@ class UserRepository(RedisAdapter): from python_repositories import ( ConnectionAwareInterface, ContextAwareInterface, + FileBackedQueueAdapter, + FileQueueConfig, JsonRepositoryInterface, + MemoryQueueAdapter, MinioAdapter, MinioConfig, ObjectRepositoryInterface, PostgresAdapter, PostgresConfig, + QueueRepositoryInterface, RedisAdapter, RedisConfig, TableRepositoryInterface, diff --git a/python_repositories/__init__.py b/python_repositories/__init__.py index c5124d2..cc8cca9 100644 --- a/python_repositories/__init__.py +++ b/python_repositories/__init__.py @@ -6,17 +6,28 @@ from typing import TYPE_CHECKING # Interfaces are always available; they have no optional backend dependencies. from . import adapters -from .config import MinioConfig, PostgresConfig, RedisConfig, load_dotenv +from .config import ( + FileQueueConfig, + MinioConfig, + PostgresConfig, + RedisConfig, + load_dotenv, +) from .interfaces import ( ConnectionAwareInterface, ContextAwareInterface, JsonRepositoryInterface, ObjectRepositoryInterface, + QueueRepositoryInterface, TableRepositoryInterface, ) # Adapters are imported only for static type checkers; runtime loading is delegated below. if TYPE_CHECKING: + from .adapters.file_backed_queue_adapter import ( + FileBackedQueueAdapter as FileBackedQueueAdapter, + ) + from .adapters.memory_queue_adapter import MemoryQueueAdapter as MemoryQueueAdapter from .adapters.minio_adapter import MinioAdapter as MinioAdapter from .adapters.postgres_adapter import PostgresAdapter as PostgresAdapter from .adapters.redis_adapter import RedisAdapter as RedisAdapter @@ -24,10 +35,12 @@ if TYPE_CHECKING: __all__ = [ "ConnectionAwareInterface", "ContextAwareInterface", + "FileQueueConfig", "JsonRepositoryInterface", "MinioConfig", "ObjectRepositoryInterface", "PostgresConfig", + "QueueRepositoryInterface", "RedisConfig", "TableRepositoryInterface", "load_dotenv", diff --git a/python_repositories/adapters/__init__.py b/python_repositories/adapters/__init__.py index 995d1ff..acdaad9 100644 --- a/python_repositories/adapters/__init__.py +++ b/python_repositories/adapters/__init__.py @@ -11,6 +11,8 @@ from typing import TYPE_CHECKING # Adapters are imported only for static type checkers; runtime loading is deferred below. if TYPE_CHECKING: + from .file_backed_queue_adapter import FileBackedQueueAdapter + from .memory_queue_adapter import MemoryQueueAdapter from .minio_adapter import MinioAdapter from .postgres_adapter import PostgresAdapter from .redis_adapter import RedisAdapter @@ -22,12 +24,19 @@ _LAZY_EXPORTS = { "RedisAdapter": (".redis_adapter", "RedisAdapter"), "MinioAdapter": (".minio_adapter", "MinioAdapter"), "PostgresAdapter": (".postgres_adapter", "PostgresAdapter"), + "MemoryQueueAdapter": (".memory_queue_adapter", "MemoryQueueAdapter"), + "FileBackedQueueAdapter": ( + ".file_backed_queue_adapter", + "FileBackedQueueAdapter", + ), } __all__ = [ "RedisAdapter", "MinioAdapter", "PostgresAdapter", + "MemoryQueueAdapter", + "FileBackedQueueAdapter", ] diff --git a/python_repositories/adapters/file_backed_queue_adapter.py b/python_repositories/adapters/file_backed_queue_adapter.py new file mode 100644 index 0000000..243789d --- /dev/null +++ b/python_repositories/adapters/file_backed_queue_adapter.py @@ -0,0 +1,220 @@ +"""File-backed queue adapter: in-memory hot buffer mirrored to JSONL on disk.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +import json +import os +from pathlib import Path +import threading +from typing import Any, Self, TextIO + +import structlog + +from python_repositories.adapters.memory_queue_adapter import ( + MemoryQueueAdapter, + _parse_age, +) +from python_repositories.config.file_queue_config import FileQueueConfig +from python_repositories.interfaces.queue_repository_interface import ( + QueueRepositoryInterface, +) + + +def _json_default(value: object) -> str: + if isinstance(value, datetime): + return value.isoformat() + raise TypeError(f"Object of type {type(value)!r} is not JSON serializable") + + +class FileBackedQueueAdapter(QueueRepositoryInterface): + """Queue that mirrors an in-memory buffer to an append-friendly JSONL file. + + Does not extend ``ConnectionAwareAdapter``. ``connect()`` replays the JSONL + file into memory; ``disconnect()`` flushes any open handle. + """ + + def __init__( + self, + *, + config: FileQueueConfig | None = None, + memory: MemoryQueueAdapter | None = None, + ) -> None: + if config is None: + config = FileQueueConfig.from_env() + if memory is None: + memory = MemoryQueueAdapter( + dedup_keys=config.dedup_keys, + age_key=config.age_key, + ) + self._config = config + self._memory = memory + self._lock = threading.RLock() + self._connected = False + self._append_handle: TextIO | None = None + self.logger = structlog.get_logger(self.__class__.__name__) + + def __enter__(self) -> Self: + self.connect() + return self + + def __exit__( + self, exc_type: type | None, exc_val: object | None, exc_tb: object | None + ) -> None: + del exc_type, exc_val, exc_tb + self.disconnect() + + def connect(self) -> None: + """Create parent directories and replay the JSONL file into memory.""" + with self._lock: + if self._connected: + return + path = self._config.path + path.parent.mkdir(parents=True, exist_ok=True) + self._memory.clear() + if path.is_file(): + self._replay_file(path) + self._append_handle = path.open("a", encoding="utf-8") + self._connected = True + + def disconnect(self) -> None: + """Flush and close the append handle if open.""" + with self._lock: + if self._append_handle is not None: + self._append_handle.flush() + try: + os.fsync(self._append_handle.fileno()) + except OSError: + pass + self._append_handle.close() + self._append_handle = None + self._connected = False + + def is_connected(self) -> bool: + """Return whether ``connect()`` has completed successfully.""" + return self._connected + + def enqueue(self, items: list[dict[str, Any]]) -> None: + """Append items to memory and the JSONL file; optionally age-evict.""" + self._require_connected() + with self._lock: + added = self._memory.enqueue_and_return_added(items) + if added: + self._append_items(added) + if self._config.age_key is not None: + cutoff = datetime.now(UTC) - timedelta( + hours=self._config.max_age_hours + ) + if self._memory.evict_older_than(cutoff): + self._compact() + + def dequeue_batch(self, *, max_items: int = 1000) -> list[dict[str, Any]]: + """Dequeue from memory and compact the JSONL file to match.""" + self._require_connected() + with self._lock: + batch = self._memory.dequeue_batch(max_items=max_items) + if batch: + self._compact() + return batch + + def size(self) -> int: + """Return the number of items currently in the in-memory buffer.""" + self._require_connected() + return self._memory.size() + + def evict_older_than(self, cutoff: datetime) -> int: + """Evict aged items from memory and compact the JSONL file.""" + self._require_connected() + with self._lock: + removed = self._memory.evict_older_than(cutoff) + if removed: + self._compact() + return removed + + def _require_connected(self) -> None: + if not self._connected: + raise RuntimeError("FileBackedQueueAdapter is not connected") + + def _replay_file(self, path: Path) -> None: + cutoff: datetime | None = None + if self._config.age_key is not None: + cutoff = datetime.now(UTC) - timedelta(hours=self._config.max_age_hours) + with path.open(encoding="utf-8") as handle: + for line_number, line in enumerate(handle, start=1): + stripped = line.strip() + if not stripped: + continue + try: + payload = json.loads(stripped) + except json.JSONDecodeError: + self.logger.warning( + "Skipping corrupt JSONL line", + path=str(path), + line_number=line_number, + ) + continue + if not isinstance(payload, dict): + self.logger.warning( + "Skipping non-object JSONL line", + path=str(path), + line_number=line_number, + ) + continue + item: dict[str, Any] = payload + if cutoff is not None and self._config.age_key is not None: + age_value = item.get(self._config.age_key) + if age_value is None: + self.logger.warning( + "Skipping item missing age key on replay", + path=str(path), + line_number=line_number, + age_key=self._config.age_key, + ) + continue + try: + if _parse_age(age_value) < cutoff: + continue + except ValueError: + self.logger.warning( + "Skipping item with invalid age on replay", + path=str(path), + line_number=line_number, + ) + continue + try: + self._memory.enqueue_and_return_added([item]) + except ValueError as exc: + self.logger.warning( + "Skipping invalid item on replay", + path=str(path), + line_number=line_number, + error=str(exc), + ) + + def _append_items(self, items: list[dict[str, Any]]) -> None: + if self._append_handle is None: + raise RuntimeError("append handle is not open") + for item in items: + self._append_handle.write( + json.dumps(item, default=_json_default, separators=(",", ":")) + ) + self._append_handle.write("\n") + self._append_handle.flush() + + def _compact(self) -> None: + """Rewrite the JSONL file from the current in-memory snapshot.""" + path = self._config.path + tmp_path = path.with_suffix(path.suffix + ".tmp") + if self._append_handle is not None: + self._append_handle.flush() + self._append_handle.close() + self._append_handle = None + with tmp_path.open("w", encoding="utf-8") as handle: + for item in self._memory.snapshot(): + handle.write( + json.dumps(item, default=_json_default, separators=(",", ":")) + ) + handle.write("\n") + handle.flush() + tmp_path.replace(path) + self._append_handle = path.open("a", encoding="utf-8") diff --git a/python_repositories/adapters/memory_queue_adapter.py b/python_repositories/adapters/memory_queue_adapter.py new file mode 100644 index 0000000..b5eab1f --- /dev/null +++ b/python_repositories/adapters/memory_queue_adapter.py @@ -0,0 +1,126 @@ +"""In-memory queue adapter with optional dedup and age-based eviction.""" + +from __future__ import annotations + +from collections import OrderedDict +from collections.abc import Hashable +from datetime import UTC, datetime +import threading +from typing import Any + +from python_repositories.interfaces.queue_repository_interface import ( + QueueRepositoryInterface, +) + + +def _parse_age(value: object) -> datetime: + """Normalize an age field to a timezone-aware UTC datetime.""" + if isinstance(value, datetime): + if value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) + if isinstance(value, str): + normalized = value.replace("Z", "+00:00") + parsed = datetime.fromisoformat(normalized) + if parsed.tzinfo is None: + return parsed.replace(tzinfo=UTC) + return parsed.astimezone(UTC) + raise ValueError(f"age value must be datetime or ISO-8601 str, got {type(value)!r}") + + +class MemoryQueueAdapter(QueueRepositoryInterface): + """Thread-safe in-process FIFO queue with optional deduplication.""" + + def __init__( + self, + *, + dedup_keys: tuple[str, ...] = (), + age_key: str | None = None, + ) -> None: + self._dedup_keys = dedup_keys + self._age_key = age_key + self._lock = threading.RLock() + self._items: OrderedDict[Hashable, dict[str, Any]] = OrderedDict() + self._seq = 0 + + def enqueue(self, items: list[dict[str, Any]]) -> None: + """Append items; skip duplicates when ``dedup_keys`` is configured.""" + self.enqueue_and_return_added(items) + + def enqueue_and_return_added( + self, items: list[dict[str, Any]] + ) -> list[dict[str, Any]]: + """Enqueue items and return the subset that was newly stored.""" + if not items: + return [] + added: list[dict[str, Any]] = [] + with self._lock: + for raw in items: + item = self._normalize_item(raw) + key = self._make_key(item) + if self._dedup_keys and key in self._items: + continue + self._items[key] = item + added.append(item) + return added + + def dequeue_batch(self, *, max_items: int = 1000) -> list[dict[str, Any]]: + """Remove and return up to ``max_items`` items in FIFO order.""" + if max_items < 0: + raise ValueError("max_items must be >= 0") + with self._lock: + batch: list[dict[str, Any]] = [] + for _ in range(min(max_items, len(self._items))): + _key, item = self._items.popitem(last=False) + batch.append(item) + return batch + + def size(self) -> int: + """Return the number of items currently in the queue.""" + with self._lock: + return len(self._items) + + def evict_older_than(self, cutoff: datetime) -> int: + """Remove items whose age field is strictly older than ``cutoff``.""" + if self._age_key is None: + return 0 + cutoff_utc = _parse_age(cutoff) + removed = 0 + with self._lock: + to_remove = [ + key + for key, item in self._items.items() + if _parse_age(item[self._age_key]) < cutoff_utc + ] + for key in to_remove: + del self._items[key] + removed += 1 + return removed + + def clear(self) -> None: + """Remove all items from the queue.""" + with self._lock: + self._items.clear() + + def snapshot(self) -> list[dict[str, Any]]: + """Return a shallow copy of queued items in FIFO order.""" + with self._lock: + return [dict(item) for item in self._items.values()] + + def _normalize_item(self, raw: dict[str, Any]) -> dict[str, Any]: + item = dict(raw) + if self._dedup_keys: + missing = [key for key in self._dedup_keys if key not in item] + if missing: + raise ValueError(f"item missing dedup key(s): {missing}") + if self._age_key is not None: + if self._age_key not in item: + raise ValueError(f"item missing age key: {self._age_key!r}") + item[self._age_key] = _parse_age(item[self._age_key]) + return item + + def _make_key(self, item: dict[str, Any]) -> Hashable: + if not self._dedup_keys: + self._seq += 1 + return self._seq + return tuple(item[key] for key in self._dedup_keys) diff --git a/python_repositories/config/__init__.py b/python_repositories/config/__init__.py index 1465415..3caf5b7 100644 --- a/python_repositories/config/__init__.py +++ b/python_repositories/config/__init__.py @@ -1,9 +1,11 @@ from .dotenv_loader import load_dotenv as load_dotenv +from .file_queue_config import FileQueueConfig as FileQueueConfig from .minio_config import MinioConfig as MinioConfig from .postgres_config import PostgresConfig as PostgresConfig from .redis_config import RedisConfig as RedisConfig __all__ = [ + "FileQueueConfig", "MinioConfig", "PostgresConfig", "RedisConfig", diff --git a/python_repositories/config/file_queue_config.py b/python_repositories/config/file_queue_config.py new file mode 100644 index 0000000..2a2d692 --- /dev/null +++ b/python_repositories/config/file_queue_config.py @@ -0,0 +1,56 @@ +"""File-backed queue configuration.""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +from pathlib import Path + +from python_utils import check_env + +from python_repositories.config.dotenv_loader import load_dotenv + +_DEFAULT_MAX_AGE_HOURS = 24 + + +@dataclass(frozen=True) +class FileQueueConfig: + """Configuration for a JSONL file-backed queue. + + Path and key schema are owned by the caller. This package does not assume + any particular directory layout or item field names. + """ + + path: Path + max_age_hours: int = _DEFAULT_MAX_AGE_HOURS + dedup_keys: tuple[str, ...] = () + age_key: str | None = None + + @classmethod + def from_env( + cls, + path_env_var_name: str = "FILE_QUEUE_PATH", + *, + max_age_hours_env_var_name: str = "FILE_QUEUE_MAX_AGE_HOURS", + dedup_keys: tuple[str, ...] = (), + age_key: str | None = None, + use_dotenv: bool = True, + ) -> FileQueueConfig: + """Load path and optional max age from environment variables. + + ``dedup_keys`` and ``age_key`` are domain-specific and must be passed + explicitly; they are not read from the environment. + """ + if use_dotenv: + load_dotenv() + check_env(path_env_var_name) + raw_max_age = os.getenv(max_age_hours_env_var_name) + max_age_hours = ( + _DEFAULT_MAX_AGE_HOURS if raw_max_age is None else int(raw_max_age) + ) + return cls( + path=Path(str(os.getenv(path_env_var_name))), + max_age_hours=max_age_hours, + dedup_keys=dedup_keys, + age_key=age_key, + ) diff --git a/python_repositories/interfaces/__init__.py b/python_repositories/interfaces/__init__.py index 2135084..c4d439f 100644 --- a/python_repositories/interfaces/__init__.py +++ b/python_repositories/interfaces/__init__.py @@ -8,6 +8,9 @@ from .json_repository_interface import ( from .object_repository_interface import ( ObjectRepositoryInterface as ObjectRepositoryInterface, ) +from .queue_repository_interface import ( + QueueRepositoryInterface as QueueRepositoryInterface, +) from .table_repository_interface import ( TableRepositoryInterface as TableRepositoryInterface, ) @@ -17,5 +20,6 @@ __all__ = [ "ContextAwareInterface", "JsonRepositoryInterface", "ObjectRepositoryInterface", + "QueueRepositoryInterface", "TableRepositoryInterface", ] diff --git a/python_repositories/interfaces/queue_repository_interface.py b/python_repositories/interfaces/queue_repository_interface.py new file mode 100644 index 0000000..11eb340 --- /dev/null +++ b/python_repositories/interfaces/queue_repository_interface.py @@ -0,0 +1,41 @@ +"""Definition of QueueRepositoryInterface protocol.""" + +from abc import abstractmethod +from datetime import datetime +from typing import Any, Protocol, runtime_checkable + + +@runtime_checkable +class QueueRepositoryInterface(Protocol): + """Interface that defines buffered queue enqueue/dequeue methods.""" + + @abstractmethod + def enqueue(self, items: list[dict[str, Any]]) -> None: + """Append items to the queue. + + Duplicates may be skipped when the implementation is configured with + dedup keys. An empty list is a no-op. + """ + ... + + @abstractmethod + def dequeue_batch(self, *, max_items: int = 1000) -> list[dict[str, Any]]: + """Remove and return up to ``max_items`` items in FIFO order. + + Returns an empty list when the queue is empty. + """ + ... + + @abstractmethod + def size(self) -> int: + """Return the number of items currently in the queue.""" + ... + + @abstractmethod + def evict_older_than(self, cutoff: datetime) -> int: + """Remove items older than ``cutoff`` and return how many were removed. + + Age is determined by an implementation-specific item field. When no age + field is configured, this is a no-op that returns ``0``. + """ + ... diff --git a/tests/unit/adapters_test.py b/tests/unit/adapters_test.py index 164c8fe..f02d955 100644 --- a/tests/unit/adapters_test.py +++ b/tests/unit/adapters_test.py @@ -46,6 +46,8 @@ assert JsonRepositoryInterface is not None assert "python_repositories.adapters.redis_adapter" not in sys.modules assert "python_repositories.adapters.minio_adapter" not in sys.modules assert "python_repositories.adapters.postgres_adapter" not in sys.modules +assert "python_repositories.adapters.memory_queue_adapter" not in sys.modules +assert "python_repositories.adapters.file_backed_queue_adapter" not in sys.modules """ result = subprocess.run( [sys.executable, "-c", script], @@ -59,11 +61,19 @@ assert "python_repositories.adapters.postgres_adapter" not in sys.modules def test_lazy_adapter_load_succeeds_when_extra_present() -> None: """Adapters load when their optional dependencies are installed.""" - from python_repositories import MinioAdapter, PostgresAdapter, RedisAdapter + from python_repositories import ( + FileBackedQueueAdapter, + MemoryQueueAdapter, + MinioAdapter, + PostgresAdapter, + RedisAdapter, + ) assert RedisAdapter.__name__ == "RedisAdapter" assert MinioAdapter.__name__ == "MinioAdapter" assert PostgresAdapter.__name__ == "PostgresAdapter" + assert MemoryQueueAdapter.__name__ == "MemoryQueueAdapter" + assert FileBackedQueueAdapter.__name__ == "FileBackedQueueAdapter" def test_redis_adapter_import_error_without_extra() -> None: @@ -149,9 +159,13 @@ def test_adapters_dir_exposes_lazy_exports() -> None: """dir(adapters) includes lazy adapter names for tab completion.""" import python_repositories.adapters as adapters - assert {"RedisAdapter", "MinioAdapter", "PostgresAdapter"}.issubset( - set(dir(adapters)) - ) + assert { + "RedisAdapter", + "MinioAdapter", + "PostgresAdapter", + "MemoryQueueAdapter", + "FileBackedQueueAdapter", + }.issubset(set(dir(adapters))) def test_adapters_getattr_raises_for_unknown() -> None: @@ -167,3 +181,5 @@ def test_top_level_dir_exposes_lazy_exports() -> None: assert "RedisAdapter" in dir(python_repositories) assert "MinioAdapter" in dir(python_repositories) assert "PostgresAdapter" in dir(python_repositories) + assert "MemoryQueueAdapter" in dir(python_repositories) + assert "FileBackedQueueAdapter" in dir(python_repositories) diff --git a/tests/unit/file_backed_queue_adapter_test.py b/tests/unit/file_backed_queue_adapter_test.py new file mode 100644 index 0000000..2972da8 --- /dev/null +++ b/tests/unit/file_backed_queue_adapter_test.py @@ -0,0 +1,241 @@ +"""Unit tests for FileBackedQueueAdapter.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest + +from python_repositories.adapters.file_backed_queue_adapter import ( + FileBackedQueueAdapter, +) +from python_repositories.adapters.memory_queue_adapter import MemoryQueueAdapter +from python_repositories.config.file_queue_config import FileQueueConfig +from python_repositories.interfaces.queue_repository_interface import ( + QueueRepositoryInterface, +) + + +def _config(path: Path, **kwargs: object) -> FileQueueConfig: + return FileQueueConfig(path=path, **kwargs) # type: ignore[arg-type] + + +def test_implements_interface() -> None: + assert issubclass(FileBackedQueueAdapter, QueueRepositoryInterface) + + +def test_ops_require_connect(tmp_path: Path) -> None: + queue = FileBackedQueueAdapter(config=_config(tmp_path / "q.jsonl")) + with pytest.raises(RuntimeError, match="not connected"): + queue.enqueue([{"id": 1}]) + with pytest.raises(RuntimeError, match="not connected"): + queue.dequeue_batch() + with pytest.raises(RuntimeError, match="not connected"): + queue.size() + with pytest.raises(RuntimeError, match="not connected"): + queue.evict_older_than(datetime.now(UTC)) + + +def test_enqueue_dequeue_persists_and_compacts(tmp_path: Path) -> None: + path = tmp_path / "items.jsonl" + with FileBackedQueueAdapter(config=_config(path, dedup_keys=("id",))) as queue: + queue.enqueue([{"id": 1}, {"id": 2}, {"id": 1}]) + assert queue.size() == 2 + assert path.is_file() + lines = path.read_text(encoding="utf-8").strip().splitlines() + assert len(lines) == 2 + batch = queue.dequeue_batch(max_items=1) + assert batch == [{"id": 1}] + remaining = path.read_text(encoding="utf-8").strip().splitlines() + assert len(remaining) == 1 + assert queue.size() == 1 + + +def test_replay_on_connect(tmp_path: Path) -> None: + path = tmp_path / "items.jsonl" + config = _config(path, dedup_keys=("id",), age_key="created_at", max_age_hours=24) + now = datetime.now(UTC) + with FileBackedQueueAdapter(config=config) as queue: + queue.enqueue( + [ + {"id": 1, "created_at": now - timedelta(hours=1)}, + {"id": 2, "created_at": now - timedelta(minutes=10)}, + ] + ) + with FileBackedQueueAdapter(config=config) as queue: + assert queue.size() == 2 + assert [item["id"] for item in queue.dequeue_batch(max_items=10)] == [1, 2] + + +def test_replay_filters_by_max_age(tmp_path: Path) -> None: + path = tmp_path / "items.jsonl" + now = datetime.now(UTC) + path.write_text( + "\n".join( + [ + f'{{"id": "old", "created_at": "{(now - timedelta(hours=30)).isoformat()}"}}', + f'{{"id": "new", "created_at": "{(now - timedelta(hours=1)).isoformat()}"}}', + ] + ) + + "\n", + encoding="utf-8", + ) + config = _config(path, dedup_keys=("id",), age_key="created_at", max_age_hours=24) + with FileBackedQueueAdapter(config=config) as queue: + assert queue.size() == 1 + assert queue.dequeue_batch(max_items=10)[0]["id"] == "new" + + +def test_connect_skips_corrupt_and_invalid_lines(tmp_path: Path) -> None: + path = tmp_path / "items.jsonl" + path.write_text( + "\n".join( + [ + "", + "not-json", + "[1, 2]", + '{"id": 1}', + '{"name": "missing-id"}', + ] + ) + + "\n", + encoding="utf-8", + ) + config = _config(path, dedup_keys=("id",)) + with FileBackedQueueAdapter(config=config) as queue: + assert queue.size() == 1 + assert queue.dequeue_batch(max_items=10) == [{"id": 1}] + + +def test_connect_skips_missing_or_invalid_age_on_replay(tmp_path: Path) -> None: + path = tmp_path / "items.jsonl" + now = datetime.now(UTC) + path.write_text( + "\n".join( + [ + '{"id": 1}', + '{"id": 2, "created_at": "not-a-date"}', + f'{{"id": 3, "created_at": "{now.isoformat()}"}}', + ] + ) + + "\n", + encoding="utf-8", + ) + config = _config(path, dedup_keys=("id",), age_key="created_at") + with FileBackedQueueAdapter(config=config) as queue: + assert queue.size() == 1 + assert queue.dequeue_batch(max_items=10)[0]["id"] == 3 + + +def test_connect_is_idempotent(tmp_path: Path) -> None: + path = tmp_path / "items.jsonl" + queue = FileBackedQueueAdapter(config=_config(path)) + queue.connect() + queue.enqueue([{"id": 1}]) + queue.connect() + assert queue.size() == 1 + queue.disconnect() + assert not queue.is_connected() + + +def test_evict_compacts_file(tmp_path: Path) -> None: + path = tmp_path / "items.jsonl" + config = _config(path, dedup_keys=("id",), age_key="created_at", max_age_hours=24) + now = datetime.now(UTC) + with FileBackedQueueAdapter(config=config) as queue: + queue.enqueue( + [ + {"id": 1, "created_at": now - timedelta(hours=2)}, + {"id": 2, "created_at": now - timedelta(minutes=5)}, + ] + ) + removed = queue.evict_older_than(now - timedelta(hours=1)) + assert removed == 1 + assert queue.size() == 1 + lines = [line for line in path.read_text(encoding="utf-8").splitlines() if line] + assert len(lines) == 1 + + +def test_auto_evict_on_enqueue(tmp_path: Path) -> None: + path = tmp_path / "items.jsonl" + config = _config(path, dedup_keys=("id",), age_key="created_at", max_age_hours=1) + now = datetime.now(UTC) + with FileBackedQueueAdapter(config=config) as queue: + queue.enqueue([{"id": 1, "created_at": now - timedelta(hours=2)}]) + # Item is enqueued then immediately age-evicted. + assert queue.size() == 0 + + +def test_from_env_when_config_omitted( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + path = tmp_path / "from-env.jsonl" + monkeypatch.setenv("FILE_QUEUE_PATH", str(path)) + with FileBackedQueueAdapter() as queue: + queue.enqueue([{"id": 1}]) + assert queue.size() == 1 + + +def test_injected_memory(tmp_path: Path) -> None: + path = tmp_path / "items.jsonl" + memory = MemoryQueueAdapter(dedup_keys=("id",)) + config = _config(path, dedup_keys=("id",)) + with FileBackedQueueAdapter(config=config, memory=memory) as queue: + queue.enqueue([{"id": 1}]) + assert memory.size() == 1 + + +def test_empty_enqueue_and_dequeue_skip_compact(tmp_path: Path) -> None: + path = tmp_path / "items.jsonl" + with FileBackedQueueAdapter(config=_config(path)) as queue: + queue.enqueue([]) + assert queue.dequeue_batch(max_items=10) == [] + assert queue.evict_older_than(datetime.now(UTC)) == 0 + assert not path.exists() or path.read_text(encoding="utf-8") == "" + + +def test_reconnect_replays_after_disconnect(tmp_path: Path) -> None: + path = tmp_path / "items.jsonl" + queue = FileBackedQueueAdapter(config=_config(path, dedup_keys=("id",))) + queue.connect() + queue.enqueue([{"id": 1}]) + queue.disconnect() + queue.connect() + assert queue.size() == 1 + queue.disconnect() + + +def test_json_default_rejects_unsupported() -> None: + from python_repositories.adapters.file_backed_queue_adapter import _json_default + + with pytest.raises(TypeError): + _json_default(object()) + + +def test_disconnect_ignores_fsync_oserror( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + path = tmp_path / "items.jsonl" + queue = FileBackedQueueAdapter(config=_config(path)) + queue.connect() + + def boom(_fd: int) -> None: + raise OSError("fsync failed") + + monkeypatch.setattr( + "python_repositories.adapters.file_backed_queue_adapter.os.fsync", + boom, + ) + queue.disconnect() + assert not queue.is_connected() + + +def test_append_items_requires_open_handle(tmp_path: Path) -> None: + path = tmp_path / "items.jsonl" + queue = FileBackedQueueAdapter(config=_config(path)) + queue.connect() + queue._append_handle = None # noqa: SLF001 + with pytest.raises(RuntimeError, match="append handle is not open"): + queue._append_items([{"id": 1}]) # noqa: SLF001 + queue._connected = False # noqa: SLF001 diff --git a/tests/unit/file_queue_config_test.py b/tests/unit/file_queue_config_test.py new file mode 100644 index 0000000..722d86c --- /dev/null +++ b/tests/unit/file_queue_config_test.py @@ -0,0 +1,56 @@ +"""Unit tests for FileQueueConfig.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from python_repositories.config import FileQueueConfig + + +def test_defaults() -> None: + config = FileQueueConfig(path=Path("/tmp/queue.jsonl")) + assert config.path == Path("/tmp/queue.jsonl") + assert config.max_age_hours == 24 + assert config.dedup_keys == () + assert config.age_key is None + + +def test_from_env_loads_path(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("FILE_QUEUE_PATH", "/data/items.jsonl") + monkeypatch.delenv("FILE_QUEUE_MAX_AGE_HOURS", raising=False) + config = FileQueueConfig.from_env(use_dotenv=False) + assert config.path == Path("/data/items.jsonl") + assert config.max_age_hours == 24 + + +def test_from_env_loads_max_age_hours(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("FILE_QUEUE_PATH", "/data/items.jsonl") + monkeypatch.setenv("FILE_QUEUE_MAX_AGE_HOURS", "48") + config = FileQueueConfig.from_env(use_dotenv=False) + assert config.max_age_hours == 48 + + +def test_from_env_raises_when_path_missing(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("FILE_QUEUE_PATH", raising=False) + with pytest.raises(Exception): + FileQueueConfig.from_env(use_dotenv=False) + + +def test_from_env_respects_custom_env_var_names( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("CUSTOM_QUEUE_PATH", "/custom/q.jsonl") + monkeypatch.setenv("CUSTOM_MAX_AGE", "12") + config = FileQueueConfig.from_env( + "CUSTOM_QUEUE_PATH", + max_age_hours_env_var_name="CUSTOM_MAX_AGE", + dedup_keys=("id",), + age_key="created_at", + use_dotenv=False, + ) + assert config.path == Path("/custom/q.jsonl") + assert config.max_age_hours == 12 + assert config.dedup_keys == ("id",) + assert config.age_key == "created_at" diff --git a/tests/unit/memory_queue_adapter_test.py b/tests/unit/memory_queue_adapter_test.py new file mode 100644 index 0000000..1a08c18 --- /dev/null +++ b/tests/unit/memory_queue_adapter_test.py @@ -0,0 +1,144 @@ +"""Unit tests for MemoryQueueAdapter.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +import threading + +import pytest + +from python_repositories.adapters.memory_queue_adapter import ( + MemoryQueueAdapter, + _parse_age, +) +from python_repositories.interfaces.queue_repository_interface import ( + QueueRepositoryInterface, +) + + +def test_implements_interface() -> None: + assert issubclass(MemoryQueueAdapter, QueueRepositoryInterface) + + +def test_fifo_without_dedup() -> None: + queue = MemoryQueueAdapter() + queue.enqueue([{"id": 1}, {"id": 2}, {"id": 3}]) + assert queue.size() == 3 + assert queue.dequeue_batch(max_items=2) == [{"id": 1}, {"id": 2}] + assert queue.dequeue_batch(max_items=10) == [{"id": 3}] + assert queue.dequeue_batch() == [] + assert queue.size() == 0 + + +def test_empty_enqueue_is_noop() -> None: + queue = MemoryQueueAdapter() + queue.enqueue([]) + assert queue.size() == 0 + assert queue.enqueue_and_return_added([]) == [] + + +def test_dedup_keeps_first() -> None: + queue = MemoryQueueAdapter(dedup_keys=("id",)) + queue.enqueue([{"id": 1, "v": "a"}, {"id": 1, "v": "b"}, {"id": 2, "v": "c"}]) + assert queue.size() == 2 + assert queue.dequeue_batch(max_items=10) == [ + {"id": 1, "v": "a"}, + {"id": 2, "v": "c"}, + ] + + +def test_missing_dedup_key_raises() -> None: + queue = MemoryQueueAdapter(dedup_keys=("id",)) + with pytest.raises(ValueError, match="missing dedup key"): + queue.enqueue([{"name": "x"}]) + + +def test_missing_age_key_raises() -> None: + queue = MemoryQueueAdapter(age_key="created_at") + with pytest.raises(ValueError, match="missing age key"): + queue.enqueue([{"id": 1}]) + + +def test_evict_older_than() -> None: + queue = MemoryQueueAdapter(age_key="created_at", dedup_keys=("id",)) + now = datetime.now(UTC) + queue.enqueue( + [ + {"id": 1, "created_at": now - timedelta(hours=2)}, + {"id": 2, "created_at": now - timedelta(minutes=30)}, + {"id": 3, "created_at": (now - timedelta(hours=3)).isoformat()}, + ] + ) + removed = queue.evict_older_than(now - timedelta(hours=1)) + assert removed == 2 + assert queue.size() == 1 + remaining = queue.dequeue_batch(max_items=10) + assert remaining[0]["id"] == 2 + + +def test_evict_without_age_key_is_noop() -> None: + queue = MemoryQueueAdapter() + queue.enqueue([{"id": 1}]) + assert queue.evict_older_than(datetime.now(UTC)) == 0 + assert queue.size() == 1 + + +def test_parse_age_rejects_unsupported_type() -> None: + with pytest.raises(ValueError, match="age value must be"): + _parse_age(123) + + +def test_parse_age_aware_datetime() -> None: + from datetime import timezone + + eastern = timezone(timedelta(hours=-5)) + aware = datetime(2024, 1, 1, 12, 0, 0, tzinfo=eastern) + assert _parse_age(aware) == datetime(2024, 1, 1, 17, 0, 0, tzinfo=UTC) + + +def test_parse_age_naive_datetime() -> None: + naive = datetime(2024, 1, 1, 12, 0, 0) + assert _parse_age(naive) == datetime(2024, 1, 1, 12, 0, 0, tzinfo=UTC) + + +def test_parse_age_zulu_string() -> None: + assert _parse_age("2024-01-01T00:00:00Z") == datetime(2024, 1, 1, 0, 0, 0, tzinfo=UTC) + + +def test_parse_age_naive_string() -> None: + assert _parse_age("2024-01-01T00:00:00") == datetime(2024, 1, 1, 0, 0, 0, tzinfo=UTC) + + +def test_dequeue_rejects_negative_max_items() -> None: + queue = MemoryQueueAdapter() + with pytest.raises(ValueError, match="max_items"): + queue.dequeue_batch(max_items=-1) + + +def test_clear_and_snapshot() -> None: + queue = MemoryQueueAdapter() + queue.enqueue([{"id": 1}, {"id": 2}]) + assert queue.snapshot() == [{"id": 1}, {"id": 2}] + queue.clear() + assert queue.size() == 0 + assert queue.snapshot() == [] + + +def test_thread_safety_smoke() -> None: + queue = MemoryQueueAdapter(dedup_keys=("id",)) + errors: list[BaseException] = [] + + def worker(start: int) -> None: + try: + for i in range(start, start + 50): + queue.enqueue([{"id": i}]) + except BaseException as exc: # noqa: BLE001 + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(i * 50,)) for i in range(4)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert errors == [] + assert queue.size() == 200 diff --git a/tests/unit/queue_repository_interface_test.py b/tests/unit/queue_repository_interface_test.py new file mode 100644 index 0000000..601e52b --- /dev/null +++ b/tests/unit/queue_repository_interface_test.py @@ -0,0 +1,100 @@ +"""Unit tests for QueueRepositoryInterface.""" + +from datetime import UTC, datetime +from typing import Any + +import pytest + +from python_repositories.interfaces.queue_repository_interface import ( + QueueRepositoryInterface, +) + + +class InMemoryQueueRepo: + """Plain class that satisfies QueueRepositoryInterface without inheritance.""" + + def enqueue(self, items: list[dict[str, Any]]) -> None: + pass + + def dequeue_batch(self, *, max_items: int = 1000) -> list[dict[str, Any]]: + del max_items + return [] + + def size(self) -> int: + return 0 + + def evict_older_than(self, cutoff: datetime) -> int: + del cutoff + return 0 + + +def accepts_queue_repo(repo: QueueRepositoryInterface) -> None: + """Type-checking hook for QueueRepositoryInterface structural subtyping.""" + repo.size() + + +def test_instantiation_fails_when_enqueue_not_implemented() -> None: + class Incomplete(QueueRepositoryInterface): + def dequeue_batch(self, *, max_items: int = 1000) -> list[dict[str, Any]]: + return [] + + def size(self) -> int: + return 0 + + def evict_older_than(self, cutoff: datetime) -> int: + return 0 + + with pytest.raises(TypeError): + _ = Incomplete() # type: ignore + + +def test_instantiation_fails_when_dequeue_batch_not_implemented() -> None: + class Incomplete(QueueRepositoryInterface): + def enqueue(self, items: list[dict[str, Any]]) -> None: + pass + + def size(self) -> int: + return 0 + + def evict_older_than(self, cutoff: datetime) -> int: + return 0 + + with pytest.raises(TypeError): + _ = Incomplete() # type: ignore + + +def test_instantiation_fails_when_size_not_implemented() -> None: + class Incomplete(QueueRepositoryInterface): + def enqueue(self, items: list[dict[str, Any]]) -> None: + pass + + def dequeue_batch(self, *, max_items: int = 1000) -> list[dict[str, Any]]: + return [] + + def evict_older_than(self, cutoff: datetime) -> int: + return 0 + + with pytest.raises(TypeError): + _ = Incomplete() # type: ignore + + +def test_instantiation_fails_when_evict_older_than_not_implemented() -> None: + class Incomplete(QueueRepositoryInterface): + def enqueue(self, items: list[dict[str, Any]]) -> None: + pass + + def dequeue_batch(self, *, max_items: int = 1000) -> list[dict[str, Any]]: + return [] + + def size(self) -> int: + return 0 + + with pytest.raises(TypeError): + _ = Incomplete() # type: ignore + + +def test_structural_subtyping() -> None: + repo: QueueRepositoryInterface = InMemoryQueueRepo() + accepts_queue_repo(repo) + assert isinstance(repo, QueueRepositoryInterface) + assert repo.evict_older_than(datetime.now(UTC)) == 0