47 lines
1.6 KiB
Python
47 lines
1.6 KiB
Python
from django.utils import timezone
|
|
|
|
from apps.ingestion.models import IngestionError, IngestionRun
|
|
|
|
|
|
def start_ingestion_run(*, provider_namespace: str, job_type: str, triggered_by=None, context: dict | None = None) -> IngestionRun:
|
|
return IngestionRun.objects.create(
|
|
provider_namespace=provider_namespace,
|
|
job_type=job_type,
|
|
status=IngestionRun.RunStatus.RUNNING,
|
|
triggered_by=triggered_by,
|
|
started_at=timezone.now(),
|
|
context=context or {},
|
|
)
|
|
|
|
|
|
def finish_ingestion_run(*, run: IngestionRun, status: str, processed: int = 0, created: int = 0, updated: int = 0, failed: int = 0) -> IngestionRun:
|
|
run.status = status
|
|
run.records_processed = processed
|
|
run.records_created = created
|
|
run.records_updated = updated
|
|
run.records_failed = failed
|
|
run.finished_at = timezone.now()
|
|
run.save(
|
|
update_fields=[
|
|
"status",
|
|
"records_processed",
|
|
"records_created",
|
|
"records_updated",
|
|
"records_failed",
|
|
"finished_at",
|
|
]
|
|
)
|
|
return run
|
|
|
|
|
|
def log_ingestion_error(*, run: IngestionRun, message: str, provider_namespace: str, severity: str = IngestionError.Severity.ERROR, entity_type: str = "", external_id: str = "", raw_payload: dict | None = None) -> IngestionError:
|
|
return IngestionError.objects.create(
|
|
ingestion_run=run,
|
|
provider_namespace=provider_namespace,
|
|
message=message,
|
|
severity=severity,
|
|
entity_type=entity_type,
|
|
external_id=external_id,
|
|
raw_payload=raw_payload or {},
|
|
)
|