Files
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
Add QueueRepositoryInterface with memory and file-backed adapters.
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]>
2026-07-16 20:20:52 +02:00

61 lines
1.8 KiB
Python

"""python_repositories: Unified repository interfaces and adapters."""
from __future__ import annotations
from typing import TYPE_CHECKING
# Interfaces are always available; they have no optional backend dependencies.
from . import adapters
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
__all__ = [
"ConnectionAwareInterface",
"ContextAwareInterface",
"FileQueueConfig",
"JsonRepositoryInterface",
"MinioConfig",
"ObjectRepositoryInterface",
"PostgresConfig",
"QueueRepositoryInterface",
"RedisConfig",
"TableRepositoryInterface",
"load_dotenv",
*adapters.__all__,
]
def __getattr__(name: str) -> object:
"""Delegate adapter lookups to adapters; lazy loading is defined there."""
if name in adapters.__all__:
return getattr(adapters, name)
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
def __dir__() -> list[str]:
"""Expose lazy adapter names in tab completion and dir()."""
return sorted(__all__)