85 lines
2.6 KiB
Python
85 lines
2.6 KiB
Python
from datetime import timedelta
|
|
|
|
from django.db.models import Q
|
|
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,
|
|
error_summary: str = "",
|
|
) -> IngestionRun:
|
|
run.status = status
|
|
run.records_processed = processed
|
|
run.records_created = created
|
|
run.records_updated = updated
|
|
run.records_failed = failed
|
|
run.error_summary = error_summary
|
|
run.finished_at = timezone.now()
|
|
run.save(
|
|
update_fields=[
|
|
"status",
|
|
"records_processed",
|
|
"records_created",
|
|
"records_updated",
|
|
"records_failed",
|
|
"error_summary",
|
|
"finished_at",
|
|
]
|
|
)
|
|
return run
|
|
|
|
|
|
def mark_ingestion_run_skipped(*, provider_namespace: str, job_type: str, reason: str, context: dict | None = None) -> IngestionRun:
|
|
now = timezone.now()
|
|
run = IngestionRun.objects.create(
|
|
provider_namespace=provider_namespace,
|
|
job_type=job_type,
|
|
status=IngestionRun.RunStatus.CANCELED,
|
|
started_at=now,
|
|
finished_at=now,
|
|
error_summary=reason,
|
|
context=context or {},
|
|
)
|
|
return run
|
|
|
|
|
|
def has_running_ingestion_run(*, provider_namespace: str, job_type: str, within_minutes: int) -> bool:
|
|
cutoff = timezone.now() - timedelta(minutes=max(within_minutes, 1))
|
|
return IngestionRun.objects.filter(
|
|
provider_namespace=provider_namespace,
|
|
job_type=job_type,
|
|
status=IngestionRun.RunStatus.RUNNING,
|
|
started_at__gte=cutoff,
|
|
).filter(Q(finished_at__isnull=True) | Q(finished_at__gte=cutoff)).exists()
|
|
|
|
|
|
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 {},
|
|
)
|