49 lines
2.2 KiB
Python
49 lines
2.2 KiB
Python
"""Explicit direct-data launch configuration. Construction makes no hotel requests."""
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
from arr_web.arr_data_executor import DirectARRExecutor
|
|
from arr_web.arr_downloads import PersistentARRDownloads
|
|
from integrations.ohip.arr_data import ARRDataSource
|
|
from integrations.ohip.audit_arr_capture import protected_read
|
|
from integrations.ohip.capture_job import atomic_json, fingerprint, job_lock, private_directory, sync_directory
|
|
from integrations.ohip.collect_arr_source import require, strict_json
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DirectARRSource:
|
|
root: Path
|
|
hotel_id: str
|
|
credential_file: Path | None = None
|
|
transport_factory: Callable | None = None
|
|
page_size: int = 100
|
|
max_pages: int = 100
|
|
max_records: int = 10000
|
|
max_requests: int = 25000
|
|
|
|
|
|
def compose_direct_arr_downloads(source: DirectARRSource, processing):
|
|
check_schema = getattr(processing.repository, "assert_data_source_schema", None)
|
|
if callable(check_schema):
|
|
check_schema()
|
|
root = Path(source.root).absolute()
|
|
reader = ARRDataSource(root / "source", source.hotel_id, credential_file=source.credential_file,
|
|
transport_factory=source.transport_factory, page_size=source.page_size, max_pages=source.max_pages,
|
|
max_records=source.max_records, max_requests=source.max_requests)
|
|
executor = DirectARRExecutor(root=root / "processing", source=reader, policy=processing.policy,
|
|
object_store=processing.object_store, repository=processing.repository, ingestion=processing.ingestion,
|
|
processor=processing.processor)
|
|
private_directory(root)
|
|
sync_directory(root.parent)
|
|
identity = executor.source_identity()
|
|
with job_lock(root):
|
|
pin = root / "source.json"
|
|
if not pin.exists():
|
|
require(not (root / "queue").exists() and not (root / "processing").exists(),
|
|
"unbound_data_runtime_state")
|
|
atomic_json(pin, identity, replace=False)
|
|
require(fingerprint(strict_json(protected_read(pin, 65536))) == fingerprint(identity),
|
|
"data_runtime_source_conflict")
|
|
return PersistentARRDownloads(root / "queue", executor)
|