Add v2 relational domain foundations with import run/file models
This commit is contained in:
Alfredo Di Stasio
@ -1,117 +1,99 @@
|
||||
from django.contrib import admin
|
||||
from django.contrib import messages
|
||||
from django.db.models import Count
|
||||
|
||||
from apps.providers.registry import get_default_provider_namespace
|
||||
|
||||
from .models import IngestionError, IngestionRun
|
||||
from .tasks import trigger_full_sync, trigger_incremental_sync
|
||||
from .models import ImportFile, ImportRun, IngestionError, IngestionRun
|
||||
|
||||
|
||||
class IngestionErrorInline(admin.TabularInline):
|
||||
model = IngestionError
|
||||
class ImportFileInline(admin.TabularInline):
|
||||
model = ImportFile
|
||||
extra = 0
|
||||
readonly_fields = ("provider_namespace", "entity_type", "external_id", "severity", "message", "occurred_at")
|
||||
|
||||
|
||||
@admin.register(IngestionRun)
|
||||
class IngestionRunAdmin(admin.ModelAdmin):
|
||||
list_display = (
|
||||
"provider_namespace",
|
||||
"job_type",
|
||||
readonly_fields = (
|
||||
"relative_path",
|
||||
"status",
|
||||
"records_processed",
|
||||
"records_created",
|
||||
"records_updated",
|
||||
"records_failed",
|
||||
"error_count",
|
||||
"short_error_summary",
|
||||
"checksum",
|
||||
"file_size_bytes",
|
||||
"rows_total",
|
||||
"rows_upserted",
|
||||
"rows_failed",
|
||||
"error_message",
|
||||
"processed_at",
|
||||
"created_at",
|
||||
)
|
||||
|
||||
|
||||
@admin.register(ImportRun)
|
||||
class ImportRunAdmin(admin.ModelAdmin):
|
||||
list_display = (
|
||||
"id",
|
||||
"source",
|
||||
"status",
|
||||
"files_total",
|
||||
"files_processed",
|
||||
"rows_total",
|
||||
"rows_upserted",
|
||||
"rows_failed",
|
||||
"started_at",
|
||||
"finished_at",
|
||||
"created_at",
|
||||
)
|
||||
list_filter = ("provider_namespace", "job_type", "status")
|
||||
search_fields = ("provider_namespace",)
|
||||
inlines = (IngestionErrorInline,)
|
||||
list_filter = ("source", "status")
|
||||
search_fields = ("source", "error_summary")
|
||||
readonly_fields = (
|
||||
"provider_namespace",
|
||||
"job_type",
|
||||
"source",
|
||||
"status",
|
||||
"triggered_by",
|
||||
"started_at",
|
||||
"finished_at",
|
||||
"records_processed",
|
||||
"records_created",
|
||||
"records_updated",
|
||||
"records_failed",
|
||||
"files_total",
|
||||
"files_processed",
|
||||
"rows_total",
|
||||
"rows_upserted",
|
||||
"rows_failed",
|
||||
"error_summary",
|
||||
"context",
|
||||
"raw_payload",
|
||||
"created_at",
|
||||
)
|
||||
actions = (
|
||||
"enqueue_full_sync_default_provider",
|
||||
"enqueue_incremental_sync_default_provider",
|
||||
"retry_selected_runs",
|
||||
inlines = (ImportFileInline,)
|
||||
|
||||
|
||||
@admin.register(ImportFile)
|
||||
class ImportFileAdmin(admin.ModelAdmin):
|
||||
list_display = (
|
||||
"id",
|
||||
"import_run",
|
||||
"relative_path",
|
||||
"status",
|
||||
"rows_total",
|
||||
"rows_upserted",
|
||||
"rows_failed",
|
||||
"processed_at",
|
||||
)
|
||||
list_filter = ("status",)
|
||||
search_fields = ("relative_path", "checksum", "error_message")
|
||||
readonly_fields = (
|
||||
"import_run",
|
||||
"relative_path",
|
||||
"status",
|
||||
"checksum",
|
||||
"file_size_bytes",
|
||||
"rows_total",
|
||||
"rows_upserted",
|
||||
"rows_failed",
|
||||
"error_message",
|
||||
"payload_preview",
|
||||
"processed_at",
|
||||
"created_at",
|
||||
)
|
||||
|
||||
@admin.action(description="Queue full sync (default provider)")
|
||||
def enqueue_full_sync_default_provider(self, request, queryset):
|
||||
provider_namespace = get_default_provider_namespace()
|
||||
trigger_full_sync.delay(provider_namespace=provider_namespace, triggered_by_id=request.user.id)
|
||||
self.message_user(request, f"Queued full sync task for {provider_namespace}.", level=messages.SUCCESS)
|
||||
|
||||
@admin.action(description="Queue incremental sync (default provider)")
|
||||
def enqueue_incremental_sync_default_provider(self, request, queryset):
|
||||
provider_namespace = get_default_provider_namespace()
|
||||
trigger_incremental_sync.delay(provider_namespace=provider_namespace, triggered_by_id=request.user.id)
|
||||
self.message_user(request, f"Queued incremental sync task for {provider_namespace}.", level=messages.SUCCESS)
|
||||
|
||||
@admin.action(description="Retry selected ingestion runs")
|
||||
def retry_selected_runs(self, request, queryset):
|
||||
count = 0
|
||||
for run in queryset:
|
||||
if run.job_type == IngestionRun.JobType.INCREMENTAL:
|
||||
trigger_incremental_sync.delay(
|
||||
provider_namespace=run.provider_namespace,
|
||||
triggered_by_id=request.user.id,
|
||||
context={"retry_of": run.id},
|
||||
)
|
||||
else:
|
||||
trigger_full_sync.delay(
|
||||
provider_namespace=run.provider_namespace,
|
||||
triggered_by_id=request.user.id,
|
||||
context={"retry_of": run.id},
|
||||
)
|
||||
count += 1
|
||||
self.message_user(request, f"Queued {count} retry task(s).", level=messages.SUCCESS)
|
||||
|
||||
def get_queryset(self, request):
|
||||
queryset = super().get_queryset(request)
|
||||
return queryset.annotate(_error_count=Count("errors"))
|
||||
|
||||
@admin.display(ordering="_error_count", description="Errors")
|
||||
def error_count(self, obj):
|
||||
return getattr(obj, "_error_count", 0)
|
||||
|
||||
@admin.display(description="Error summary")
|
||||
def short_error_summary(self, obj):
|
||||
if not obj.error_summary:
|
||||
return "-"
|
||||
return (obj.error_summary[:90] + "...") if len(obj.error_summary) > 90 else obj.error_summary
|
||||
@admin.register(IngestionRun)
|
||||
class LegacyIngestionRunAdmin(admin.ModelAdmin):
|
||||
list_display = ("provider_namespace", "job_type", "status", "started_at", "finished_at")
|
||||
list_filter = ("provider_namespace", "job_type", "status")
|
||||
search_fields = ("provider_namespace", "error_summary")
|
||||
|
||||
|
||||
@admin.register(IngestionError)
|
||||
class IngestionErrorAdmin(admin.ModelAdmin):
|
||||
class LegacyIngestionErrorAdmin(admin.ModelAdmin):
|
||||
list_display = ("provider_namespace", "entity_type", "external_id", "severity", "occurred_at")
|
||||
list_filter = ("severity", "provider_namespace")
|
||||
search_fields = ("entity_type", "external_id", "message")
|
||||
readonly_fields = (
|
||||
"ingestion_run",
|
||||
"provider_namespace",
|
||||
"entity_type",
|
||||
"external_id",
|
||||
"severity",
|
||||
"message",
|
||||
"raw_payload",
|
||||
"occurred_at",
|
||||
)
|
||||
|
||||
@ -0,0 +1,91 @@
|
||||
# Generated by Django 5.2.12 on 2026-03-13 12:44
|
||||
|
||||
import django.db.models.deletion
|
||||
from django.conf import settings
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('ingestion', '0002_ingestionrun_error_summary'),
|
||||
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='ImportRun',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('source', models.CharField(default='snapshot', max_length=80)),
|
||||
('status', models.CharField(choices=[('pending', 'Pending'), ('running', 'Running'), ('success', 'Success'), ('failed', 'Failed'), ('canceled', 'Canceled')], default='pending', max_length=24)),
|
||||
('started_at', models.DateTimeField(blank=True, null=True)),
|
||||
('finished_at', models.DateTimeField(blank=True, null=True)),
|
||||
('files_total', models.PositiveIntegerField(default=0)),
|
||||
('files_processed', models.PositiveIntegerField(default=0)),
|
||||
('rows_total', models.PositiveIntegerField(default=0)),
|
||||
('rows_upserted', models.PositiveIntegerField(default=0)),
|
||||
('rows_failed', models.PositiveIntegerField(default=0)),
|
||||
('error_summary', models.TextField(blank=True, default='')),
|
||||
('context', models.JSONField(blank=True, default=dict)),
|
||||
('created_at', models.DateTimeField(auto_now_add=True)),
|
||||
('triggered_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='import_runs', to=settings.AUTH_USER_MODEL)),
|
||||
],
|
||||
options={
|
||||
'ordering': ['-created_at'],
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='ImportFile',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('relative_path', models.CharField(max_length=260)),
|
||||
('status', models.CharField(choices=[('pending', 'Pending'), ('processing', 'Processing'), ('success', 'Success'), ('failed', 'Failed'), ('skipped', 'Skipped')], default='pending', max_length=24)),
|
||||
('checksum', models.CharField(blank=True, max_length=128)),
|
||||
('file_size_bytes', models.PositiveBigIntegerField(blank=True, null=True)),
|
||||
('rows_total', models.PositiveIntegerField(default=0)),
|
||||
('rows_upserted', models.PositiveIntegerField(default=0)),
|
||||
('rows_failed', models.PositiveIntegerField(default=0)),
|
||||
('error_message', models.TextField(blank=True)),
|
||||
('payload_preview', models.JSONField(blank=True, default=dict)),
|
||||
('processed_at', models.DateTimeField(blank=True, null=True)),
|
||||
('created_at', models.DateTimeField(auto_now_add=True)),
|
||||
('import_run', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='files', to='ingestion.importrun')),
|
||||
],
|
||||
options={
|
||||
'ordering': ['-created_at'],
|
||||
},
|
||||
),
|
||||
migrations.AddIndex(
|
||||
model_name='importrun',
|
||||
index=models.Index(fields=['source', 'status'], name='ingestion_i_source_61db63_idx'),
|
||||
),
|
||||
migrations.AddIndex(
|
||||
model_name='importrun',
|
||||
index=models.Index(fields=['created_at'], name='ingestion_i_created_93c115_idx'),
|
||||
),
|
||||
migrations.AddIndex(
|
||||
model_name='importrun',
|
||||
index=models.Index(fields=['started_at'], name='ingestion_i_started_bf1d94_idx'),
|
||||
),
|
||||
migrations.AddIndex(
|
||||
model_name='importrun',
|
||||
index=models.Index(fields=['finished_at'], name='ingestion_i_finishe_73cbed_idx'),
|
||||
),
|
||||
migrations.AddIndex(
|
||||
model_name='importfile',
|
||||
index=models.Index(fields=['import_run', 'status'], name='ingestion_i_import__075f75_idx'),
|
||||
),
|
||||
migrations.AddIndex(
|
||||
model_name='importfile',
|
||||
index=models.Index(fields=['relative_path'], name='ingestion_i_relativ_183e34_idx'),
|
||||
),
|
||||
migrations.AddIndex(
|
||||
model_name='importfile',
|
||||
index=models.Index(fields=['processed_at'], name='ingestion_i_process_dfc080_idx'),
|
||||
),
|
||||
migrations.AddConstraint(
|
||||
model_name='importfile',
|
||||
constraint=models.UniqueConstraint(fields=('import_run', 'relative_path'), name='uq_import_file_per_run_path'),
|
||||
),
|
||||
]
|
||||
@ -2,6 +2,90 @@ from django.conf import settings
|
||||
from django.db import models
|
||||
|
||||
|
||||
class ImportRun(models.Model):
|
||||
class RunStatus(models.TextChoices):
|
||||
PENDING = "pending", "Pending"
|
||||
RUNNING = "running", "Running"
|
||||
SUCCESS = "success", "Success"
|
||||
FAILED = "failed", "Failed"
|
||||
CANCELED = "canceled", "Canceled"
|
||||
|
||||
source = models.CharField(max_length=80, default="snapshot")
|
||||
status = models.CharField(max_length=24, choices=RunStatus.choices, default=RunStatus.PENDING)
|
||||
triggered_by = models.ForeignKey(
|
||||
settings.AUTH_USER_MODEL,
|
||||
on_delete=models.SET_NULL,
|
||||
blank=True,
|
||||
null=True,
|
||||
related_name="import_runs",
|
||||
)
|
||||
started_at = models.DateTimeField(blank=True, null=True)
|
||||
finished_at = models.DateTimeField(blank=True, null=True)
|
||||
files_total = models.PositiveIntegerField(default=0)
|
||||
files_processed = models.PositiveIntegerField(default=0)
|
||||
rows_total = models.PositiveIntegerField(default=0)
|
||||
rows_upserted = models.PositiveIntegerField(default=0)
|
||||
rows_failed = models.PositiveIntegerField(default=0)
|
||||
error_summary = models.TextField(blank=True, default="")
|
||||
context = models.JSONField(default=dict, blank=True)
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
|
||||
class Meta:
|
||||
ordering = ["-created_at"]
|
||||
indexes = [
|
||||
models.Index(fields=["source", "status"]),
|
||||
models.Index(fields=["created_at"]),
|
||||
models.Index(fields=["started_at"]),
|
||||
models.Index(fields=["finished_at"]),
|
||||
]
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"{self.source} | {self.status} | {self.created_at:%Y-%m-%d %H:%M}"
|
||||
|
||||
|
||||
class ImportFile(models.Model):
|
||||
class FileStatus(models.TextChoices):
|
||||
PENDING = "pending", "Pending"
|
||||
PROCESSING = "processing", "Processing"
|
||||
SUCCESS = "success", "Success"
|
||||
FAILED = "failed", "Failed"
|
||||
SKIPPED = "skipped", "Skipped"
|
||||
|
||||
import_run = models.ForeignKey(
|
||||
"ingestion.ImportRun",
|
||||
on_delete=models.CASCADE,
|
||||
related_name="files",
|
||||
)
|
||||
relative_path = models.CharField(max_length=260)
|
||||
status = models.CharField(max_length=24, choices=FileStatus.choices, default=FileStatus.PENDING)
|
||||
checksum = models.CharField(max_length=128, blank=True)
|
||||
file_size_bytes = models.PositiveBigIntegerField(blank=True, null=True)
|
||||
rows_total = models.PositiveIntegerField(default=0)
|
||||
rows_upserted = models.PositiveIntegerField(default=0)
|
||||
rows_failed = models.PositiveIntegerField(default=0)
|
||||
error_message = models.TextField(blank=True)
|
||||
payload_preview = models.JSONField(default=dict, blank=True)
|
||||
processed_at = models.DateTimeField(blank=True, null=True)
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
|
||||
class Meta:
|
||||
ordering = ["-created_at"]
|
||||
constraints = [
|
||||
models.UniqueConstraint(
|
||||
fields=["import_run", "relative_path"],
|
||||
name="uq_import_file_per_run_path",
|
||||
),
|
||||
]
|
||||
indexes = [
|
||||
models.Index(fields=["import_run", "status"]),
|
||||
models.Index(fields=["relative_path"]),
|
||||
models.Index(fields=["processed_at"]),
|
||||
]
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"{self.relative_path} [{self.status}]"
|
||||
|
||||
|
||||
class IngestionRun(models.Model):
|
||||
class RunStatus(models.TextChoices):
|
||||
PENDING = "pending", "Pending"
|
||||
|
||||
Reference in New Issue
Block a user