32 lines
1.3 KiB
Python
32 lines
1.3 KiB
Python
from __future__ import annotations
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from .models import IngestionBatch
|
|
|
|
@dataclass
|
|
class RunResult:
|
|
run_id: str
|
|
status: str
|
|
batch: IngestionBatch | None = None
|
|
attempts: int = 0
|
|
dead_letter: list[dict] | None = None
|
|
|
|
class Orchestrator:
|
|
def __init__(self, connectors, cursor_store, publisher, max_attempts=3):
|
|
self.connectors, self.cursors, self.publisher, self.max_attempts = connectors, cursor_store, publisher, max_attempts
|
|
|
|
def run(self, connector_name, tenant_id, mode='incremental', idempotency_key=None):
|
|
connector = self.connectors[connector_name]
|
|
cursor = self.cursors.load(tenant_id, connector_name)
|
|
batch = None
|
|
for attempt in range(1, self.max_attempts + 1):
|
|
try:
|
|
batch = connector.full(tenant_id) if mode == 'full' else connector.incremental(tenant_id, cursor)
|
|
self.publisher.publish(batch)
|
|
if batch.next_cursor: self.cursors.save(batch.next_cursor)
|
|
return RunResult(str(uuid.uuid4()), 'succeeded', batch, attempt, [])
|
|
except Exception as exc:
|
|
if attempt == self.max_attempts:
|
|
return RunResult(str(uuid.uuid4()), 'dead_lettered', batch, attempt, [{'error': type(exc).__name__}])
|
|
raise AssertionError('unreachable')
|