Brian Bjarke JensenandCursor
a768474892
Provide a generic disk-backed FIFO queue with configurable path, retention, and dedup keys so consumers can buffer items across restarts without optional extras. Co-authored-by: Cursor <[email protected]>
221 lines
8.1 KiB
Python
221 lines
8.1 KiB
Python
"""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")
|