Brian Bjarke JensenandCursor
4f58c32dd6
PR Title Check / check-title (pull_request) Successful in 9s
Code Quality Pipeline / code-quality (pull_request) Failing after 53s
Test Python Package / unit-tests (pull_request) Successful in 1m1s
Test Python Package / integration-tests (pull_request) Successful in 1m44s
Test Python Package / coverage-report (pull_request) Successful in 13s
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]>
57 lines
1.6 KiB
Python
57 lines
1.6 KiB
Python
"""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,
|
|
)
|