Move queue helper functions to staticmethods on their adapters.
PR Title Check / check-title (pull_request) Successful in 8s
Test Python Package / unit-tests (pull_request) Successful in 20s
Code Quality Pipeline / code-quality (pull_request) Failing after 32s
Test Python Package / integration-tests (pull_request) Successful in 1m33s
Test Python Package / coverage-report (pull_request) Successful in 16s
PR Title Check / check-title (pull_request) Successful in 8s
Test Python Package / unit-tests (pull_request) Successful in 20s
Code Quality Pipeline / code-quality (pull_request) Failing after 32s
Test Python Package / integration-tests (pull_request) Successful in 1m33s
Test Python Package / coverage-report (pull_request) Successful in 16s
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
co-authored by
Cursor
parent
1822d886b1
commit
a78320b434
@@ -11,22 +11,13 @@ from typing import Any, Self, TextIO
|
|||||||
|
|
||||||
import structlog
|
import structlog
|
||||||
|
|
||||||
from python_repositories.adapters.memory_queue_adapter import (
|
from python_repositories.adapters.memory_queue_adapter import MemoryQueueAdapter
|
||||||
MemoryQueueAdapter,
|
|
||||||
_parse_age,
|
|
||||||
)
|
|
||||||
from python_repositories.config.file_queue_config import FileQueueConfig
|
from python_repositories.config.file_queue_config import FileQueueConfig
|
||||||
from python_repositories.interfaces.queue_repository_interface import (
|
from python_repositories.interfaces.queue_repository_interface import (
|
||||||
QueueRepositoryInterface,
|
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):
|
class FileBackedQueueAdapter(QueueRepositoryInterface):
|
||||||
"""Queue that mirrors an in-memory buffer to an append-friendly JSONL file.
|
"""Queue that mirrors an in-memory buffer to an append-friendly JSONL file.
|
||||||
|
|
||||||
@@ -54,6 +45,12 @@ class FileBackedQueueAdapter(QueueRepositoryInterface):
|
|||||||
self._append_handle: TextIO | None = None
|
self._append_handle: TextIO | None = None
|
||||||
self.logger = structlog.get_logger(self.__class__.__name__)
|
self.logger = structlog.get_logger(self.__class__.__name__)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
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")
|
||||||
|
|
||||||
def __enter__(self) -> Self:
|
def __enter__(self) -> Self:
|
||||||
self.connect()
|
self.connect()
|
||||||
return self
|
return self
|
||||||
@@ -170,7 +167,7 @@ class FileBackedQueueAdapter(QueueRepositoryInterface):
|
|||||||
)
|
)
|
||||||
continue
|
continue
|
||||||
try:
|
try:
|
||||||
if _parse_age(age_value) < cutoff:
|
if MemoryQueueAdapter._parse_age(age_value) < cutoff:
|
||||||
continue
|
continue
|
||||||
except ValueError:
|
except ValueError:
|
||||||
self.logger.warning(
|
self.logger.warning(
|
||||||
@@ -194,7 +191,7 @@ class FileBackedQueueAdapter(QueueRepositoryInterface):
|
|||||||
raise RuntimeError("append handle is not open")
|
raise RuntimeError("append handle is not open")
|
||||||
for item in items:
|
for item in items:
|
||||||
self._append_handle.write(
|
self._append_handle.write(
|
||||||
json.dumps(item, default=_json_default, separators=(",", ":"))
|
json.dumps(item, default=self._json_default, separators=(",", ":"))
|
||||||
)
|
)
|
||||||
self._append_handle.write("\n")
|
self._append_handle.write("\n")
|
||||||
self._append_handle.flush()
|
self._append_handle.flush()
|
||||||
@@ -210,7 +207,7 @@ class FileBackedQueueAdapter(QueueRepositoryInterface):
|
|||||||
with tmp_path.open("w", encoding="utf-8") as handle:
|
with tmp_path.open("w", encoding="utf-8") as handle:
|
||||||
for item in self._memory.snapshot():
|
for item in self._memory.snapshot():
|
||||||
handle.write(
|
handle.write(
|
||||||
json.dumps(item, default=_json_default, separators=(",", ":"))
|
json.dumps(item, default=self._json_default, separators=(",", ":"))
|
||||||
)
|
)
|
||||||
handle.write("\n")
|
handle.write("\n")
|
||||||
handle.flush()
|
handle.flush()
|
||||||
|
|||||||
@@ -13,21 +13,6 @@ from python_repositories.interfaces.queue_repository_interface import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
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):
|
class MemoryQueueAdapter(QueueRepositoryInterface):
|
||||||
"""Thread-safe in-process FIFO queue with optional deduplication."""
|
"""Thread-safe in-process FIFO queue with optional deduplication."""
|
||||||
|
|
||||||
@@ -43,6 +28,23 @@ class MemoryQueueAdapter(QueueRepositoryInterface):
|
|||||||
self._items: OrderedDict[Hashable, dict[str, Any]] = OrderedDict()
|
self._items: OrderedDict[Hashable, dict[str, Any]] = OrderedDict()
|
||||||
self._seq = 0
|
self._seq = 0
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
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}"
|
||||||
|
)
|
||||||
|
|
||||||
def enqueue(self, items: list[dict[str, Any]]) -> None:
|
def enqueue(self, items: list[dict[str, Any]]) -> None:
|
||||||
"""Append items; skip duplicates when ``dedup_keys`` is configured."""
|
"""Append items; skip duplicates when ``dedup_keys`` is configured."""
|
||||||
self.enqueue_and_return_added(items)
|
self.enqueue_and_return_added(items)
|
||||||
@@ -84,13 +86,13 @@ class MemoryQueueAdapter(QueueRepositoryInterface):
|
|||||||
"""Remove items whose age field is strictly older than ``cutoff``."""
|
"""Remove items whose age field is strictly older than ``cutoff``."""
|
||||||
if self._age_key is None:
|
if self._age_key is None:
|
||||||
return 0
|
return 0
|
||||||
cutoff_utc = _parse_age(cutoff)
|
cutoff_utc = self._parse_age(cutoff)
|
||||||
removed = 0
|
removed = 0
|
||||||
with self._lock:
|
with self._lock:
|
||||||
to_remove = [
|
to_remove = [
|
||||||
key
|
key
|
||||||
for key, item in self._items.items()
|
for key, item in self._items.items()
|
||||||
if _parse_age(item[self._age_key]) < cutoff_utc
|
if self._parse_age(item[self._age_key]) < cutoff_utc
|
||||||
]
|
]
|
||||||
for key in to_remove:
|
for key in to_remove:
|
||||||
del self._items[key]
|
del self._items[key]
|
||||||
@@ -116,7 +118,7 @@ class MemoryQueueAdapter(QueueRepositoryInterface):
|
|||||||
if self._age_key is not None:
|
if self._age_key is not None:
|
||||||
if self._age_key not in item:
|
if self._age_key not in item:
|
||||||
raise ValueError(f"item missing age key: {self._age_key!r}")
|
raise ValueError(f"item missing age key: {self._age_key!r}")
|
||||||
item[self._age_key] = _parse_age(item[self._age_key])
|
item[self._age_key] = self._parse_age(item[self._age_key])
|
||||||
return item
|
return item
|
||||||
|
|
||||||
def _make_key(self, item: dict[str, Any]) -> Hashable:
|
def _make_key(self, item: dict[str, Any]) -> Hashable:
|
||||||
|
|||||||
@@ -207,10 +207,8 @@ def test_reconnect_replays_after_disconnect(tmp_path: Path) -> None:
|
|||||||
|
|
||||||
|
|
||||||
def test_json_default_rejects_unsupported() -> None:
|
def test_json_default_rejects_unsupported() -> None:
|
||||||
from python_repositories.adapters.file_backed_queue_adapter import _json_default
|
|
||||||
|
|
||||||
with pytest.raises(TypeError):
|
with pytest.raises(TypeError):
|
||||||
_json_default(object())
|
FileBackedQueueAdapter._json_default(object())
|
||||||
|
|
||||||
|
|
||||||
def test_disconnect_ignores_fsync_oserror(
|
def test_disconnect_ignores_fsync_oserror(
|
||||||
|
|||||||
@@ -7,10 +7,7 @@ import threading
|
|||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from python_repositories.adapters.memory_queue_adapter import (
|
from python_repositories.adapters.memory_queue_adapter import MemoryQueueAdapter
|
||||||
MemoryQueueAdapter,
|
|
||||||
_parse_age,
|
|
||||||
)
|
|
||||||
from python_repositories.interfaces.queue_repository_interface import (
|
from python_repositories.interfaces.queue_repository_interface import (
|
||||||
QueueRepositoryInterface,
|
QueueRepositoryInterface,
|
||||||
)
|
)
|
||||||
@@ -85,7 +82,7 @@ def test_evict_without_age_key_is_noop() -> None:
|
|||||||
|
|
||||||
def test_parse_age_rejects_unsupported_type() -> None:
|
def test_parse_age_rejects_unsupported_type() -> None:
|
||||||
with pytest.raises(ValueError, match="age value must be"):
|
with pytest.raises(ValueError, match="age value must be"):
|
||||||
_parse_age(123)
|
MemoryQueueAdapter._parse_age(123)
|
||||||
|
|
||||||
|
|
||||||
def test_parse_age_aware_datetime() -> None:
|
def test_parse_age_aware_datetime() -> None:
|
||||||
@@ -93,22 +90,26 @@ def test_parse_age_aware_datetime() -> None:
|
|||||||
|
|
||||||
eastern = timezone(timedelta(hours=-5))
|
eastern = timezone(timedelta(hours=-5))
|
||||||
aware = datetime(2024, 1, 1, 12, 0, 0, tzinfo=eastern)
|
aware = datetime(2024, 1, 1, 12, 0, 0, tzinfo=eastern)
|
||||||
assert _parse_age(aware) == datetime(2024, 1, 1, 17, 0, 0, tzinfo=UTC)
|
assert MemoryQueueAdapter._parse_age(aware) == datetime(
|
||||||
|
2024, 1, 1, 17, 0, 0, tzinfo=UTC
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_parse_age_naive_datetime() -> None:
|
def test_parse_age_naive_datetime() -> None:
|
||||||
naive = datetime(2024, 1, 1, 12, 0, 0)
|
naive = datetime(2024, 1, 1, 12, 0, 0)
|
||||||
assert _parse_age(naive) == datetime(2024, 1, 1, 12, 0, 0, tzinfo=UTC)
|
assert MemoryQueueAdapter._parse_age(naive) == datetime(
|
||||||
|
2024, 1, 1, 12, 0, 0, tzinfo=UTC
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_parse_age_zulu_string() -> None:
|
def test_parse_age_zulu_string() -> None:
|
||||||
assert _parse_age("2024-01-01T00:00:00Z") == datetime(
|
assert MemoryQueueAdapter._parse_age("2024-01-01T00:00:00Z") == datetime(
|
||||||
2024, 1, 1, 0, 0, 0, tzinfo=UTC
|
2024, 1, 1, 0, 0, 0, tzinfo=UTC
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_parse_age_naive_string() -> None:
|
def test_parse_age_naive_string() -> None:
|
||||||
assert _parse_age("2024-01-01T00:00:00") == datetime(
|
assert MemoryQueueAdapter._parse_age("2024-01-01T00:00:00") == datetime(
|
||||||
2024, 1, 1, 0, 0, 0, tzinfo=UTC
|
2024, 1, 1, 0, 0, 0, tzinfo=UTC
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user