Refactor model imports and update admin classes to use pghistory for historical tracking; replace HistoricalModel with TrackedModel in relevant models

This commit is contained in:
pacnpal
2025-02-09 11:20:40 -05:00
parent 52cb51cb14
commit 64d9943d86
24 changed files with 1729 additions and 137 deletions

View File

@@ -1,16 +1,15 @@
from django.contrib import admin
from simple_history.admin import SimpleHistoryAdmin
from .models import Company, Manufacturer
@admin.register(Company)
class CompanyAdmin(SimpleHistoryAdmin):
class CompanyAdmin(admin.ModelAdmin):
list_display = ('id', 'name', 'headquarters', 'website', 'created_at')
search_fields = ('name', 'headquarters', 'description')
prepopulated_fields = {'slug': ('name',)}
readonly_fields = ('created_at', 'updated_at')
@admin.register(Manufacturer)
class ManufacturerAdmin(SimpleHistoryAdmin):
class ManufacturerAdmin(admin.ModelAdmin):
list_display = ('id', 'name', 'headquarters', 'website', 'created_at')
search_fields = ('name', 'headquarters', 'description')
prepopulated_fields = {'slug': ('name',)}

View File

@@ -0,0 +1,55 @@
# Generated by Django 5.1.4 on 2025-02-09 15:31
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("companies", "0002_add_designer_model"),
]
operations = [
migrations.CreateModel(
name="CompanyEvent",
fields=[
("pgh_id", models.AutoField(primary_key=True, serialize=False)),
("pgh_created_at", models.DateTimeField(auto_now_add=True)),
("pgh_label", models.TextField(help_text="The event label.")),
("id", models.BigIntegerField()),
("name", models.CharField(max_length=255)),
("slug", models.SlugField(db_index=False, max_length=255)),
("website", models.URLField(blank=True)),
("headquarters", models.CharField(blank=True, max_length=255)),
("description", models.TextField(blank=True)),
("total_parks", models.IntegerField(default=0)),
("total_rides", models.IntegerField(default=0)),
("created_at", models.DateTimeField(auto_now_add=True)),
("updated_at", models.DateTimeField(auto_now=True)),
],
options={
"abstract": False,
},
),
migrations.CreateModel(
name="ManufacturerEvent",
fields=[
("pgh_id", models.AutoField(primary_key=True, serialize=False)),
("pgh_created_at", models.DateTimeField(auto_now_add=True)),
("pgh_label", models.TextField(help_text="The event label.")),
("id", models.BigIntegerField()),
("name", models.CharField(max_length=255)),
("slug", models.SlugField(db_index=False, max_length=255)),
("website", models.URLField(blank=True)),
("headquarters", models.CharField(blank=True, max_length=255)),
("description", models.TextField(blank=True)),
("total_rides", models.IntegerField(default=0)),
("total_roller_coasters", models.IntegerField(default=0)),
("created_at", models.DateTimeField(auto_now_add=True)),
("updated_at", models.DateTimeField(auto_now=True)),
],
options={
"abstract": False,
},
),
]

View File

@@ -0,0 +1,131 @@
# Generated by Django 5.1.4 on 2025-02-09 15:31
import django.db.models.deletion
import pgtrigger.compiler
import pgtrigger.migrations
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("companies", "0003_companyevent_manufacturerevent"),
("pghistory", "0006_delete_aggregateevent"),
("rides", "0010_rideevent_ridemodelevent_and_more"),
]
operations = [
migrations.DeleteModel(
name="Designer",
),
migrations.AlterField(
model_name="company",
name="id",
field=models.BigAutoField(primary_key=True, serialize=False),
),
migrations.AlterField(
model_name="manufacturer",
name="id",
field=models.BigAutoField(primary_key=True, serialize=False),
),
pgtrigger.migrations.AddTrigger(
model_name="company",
trigger=pgtrigger.compiler.Trigger(
name="insert_insert",
sql=pgtrigger.compiler.UpsertTriggerSql(
func='INSERT INTO "companies_companyevent" ("created_at", "description", "headquarters", "id", "name", "pgh_context_id", "pgh_created_at", "pgh_label", "pgh_obj_id", "slug", "total_parks", "total_rides", "updated_at", "website") VALUES (NEW."created_at", NEW."description", NEW."headquarters", NEW."id", NEW."name", _pgh_attach_context(), NOW(), \'insert\', NEW."id", NEW."slug", NEW."total_parks", NEW."total_rides", NEW."updated_at", NEW."website"); RETURN NULL;',
hash="413671b13a748fb5f1acd57e8ec4af12ad7ae215",
operation="INSERT",
pgid="pgtrigger_insert_insert_a4101",
table="companies_company",
when="AFTER",
),
),
),
pgtrigger.migrations.AddTrigger(
model_name="company",
trigger=pgtrigger.compiler.Trigger(
name="update_update",
sql=pgtrigger.compiler.UpsertTriggerSql(
condition="WHEN (OLD.* IS DISTINCT FROM NEW.*)",
func='INSERT INTO "companies_companyevent" ("created_at", "description", "headquarters", "id", "name", "pgh_context_id", "pgh_created_at", "pgh_label", "pgh_obj_id", "slug", "total_parks", "total_rides", "updated_at", "website") VALUES (NEW."created_at", NEW."description", NEW."headquarters", NEW."id", NEW."name", _pgh_attach_context(), NOW(), \'update\', NEW."id", NEW."slug", NEW."total_parks", NEW."total_rides", NEW."updated_at", NEW."website"); RETURN NULL;',
hash="ee3eff1c96e46769347b8463d527668b7ece63c4",
operation="UPDATE",
pgid="pgtrigger_update_update_3d5ae",
table="companies_company",
when="AFTER",
),
),
),
pgtrigger.migrations.AddTrigger(
model_name="manufacturer",
trigger=pgtrigger.compiler.Trigger(
name="insert_insert",
sql=pgtrigger.compiler.UpsertTriggerSql(
func='INSERT INTO "companies_manufacturerevent" ("created_at", "description", "headquarters", "id", "name", "pgh_context_id", "pgh_created_at", "pgh_label", "pgh_obj_id", "slug", "total_rides", "total_roller_coasters", "updated_at", "website") VALUES (NEW."created_at", NEW."description", NEW."headquarters", NEW."id", NEW."name", _pgh_attach_context(), NOW(), \'insert\', NEW."id", NEW."slug", NEW."total_rides", NEW."total_roller_coasters", NEW."updated_at", NEW."website"); RETURN NULL;',
hash="ac3c4c31aa8dffe569154454a6c4479d189c0f64",
operation="INSERT",
pgid="pgtrigger_insert_insert_5c0b6",
table="companies_manufacturer",
when="AFTER",
),
),
),
pgtrigger.migrations.AddTrigger(
model_name="manufacturer",
trigger=pgtrigger.compiler.Trigger(
name="update_update",
sql=pgtrigger.compiler.UpsertTriggerSql(
condition="WHEN (OLD.* IS DISTINCT FROM NEW.*)",
func='INSERT INTO "companies_manufacturerevent" ("created_at", "description", "headquarters", "id", "name", "pgh_context_id", "pgh_created_at", "pgh_label", "pgh_obj_id", "slug", "total_rides", "total_roller_coasters", "updated_at", "website") VALUES (NEW."created_at", NEW."description", NEW."headquarters", NEW."id", NEW."name", _pgh_attach_context(), NOW(), \'update\', NEW."id", NEW."slug", NEW."total_rides", NEW."total_roller_coasters", NEW."updated_at", NEW."website"); RETURN NULL;',
hash="c46f36f5811cd843ff61eab3ae77624ae2e69f60",
operation="UPDATE",
pgid="pgtrigger_update_update_81971",
table="companies_manufacturer",
when="AFTER",
),
),
),
migrations.AddField(
model_name="companyevent",
name="pgh_context",
field=models.ForeignKey(
db_constraint=False,
null=True,
on_delete=django.db.models.deletion.DO_NOTHING,
related_name="+",
to="pghistory.context",
),
),
migrations.AddField(
model_name="companyevent",
name="pgh_obj",
field=models.ForeignKey(
db_constraint=False,
on_delete=django.db.models.deletion.DO_NOTHING,
related_name="events",
to="companies.company",
),
),
migrations.AddField(
model_name="manufacturerevent",
name="pgh_context",
field=models.ForeignKey(
db_constraint=False,
null=True,
on_delete=django.db.models.deletion.DO_NOTHING,
related_name="+",
to="pghistory.context",
),
),
migrations.AddField(
model_name="manufacturerevent",
name="pgh_obj",
field=models.ForeignKey(
db_constraint=False,
on_delete=django.db.models.deletion.DO_NOTHING,
related_name="events",
to="companies.manufacturer",
),
),
]

View File

@@ -2,11 +2,11 @@ from django.db import models
from django.utils.text import slugify
from django.urls import reverse
from typing import Tuple, Optional, ClassVar, TYPE_CHECKING
import pghistory
from history_tracking.models import TrackedModel, HistoricalSlug
if TYPE_CHECKING:
from history_tracking.models import HistoricalSlug
class Company(models.Model):
@pghistory.track()
class Company(TrackedModel):
name = models.CharField(max_length=255)
slug = models.SlugField(max_length=255, unique=True)
website = models.URLField(blank=True)
@@ -37,8 +37,18 @@ class Company(models.Model):
try:
return cls.objects.get(slug=slug), False
except cls.DoesNotExist:
# Check historical slugs
from history_tracking.models import HistoricalSlug
# Check pghistory first
history_model = cls.get_history_model()
history_entry = (
history_model.objects.filter(slug=slug)
.order_by('-pgh_created_at')
.first()
)
if history_entry:
return cls.objects.get(id=history_entry.pgh_obj_id), True
# Check manual slug history as fallback
try:
historical = HistoricalSlug.objects.get(
content_type__model='company',
@@ -48,7 +58,8 @@ class Company(models.Model):
except (HistoricalSlug.DoesNotExist, cls.DoesNotExist):
raise cls.DoesNotExist()
class Manufacturer(models.Model):
@pghistory.track()
class Manufacturer(TrackedModel):
name = models.CharField(max_length=255)
slug = models.SlugField(max_length=255, unique=True)
website = models.URLField(blank=True)
@@ -78,8 +89,18 @@ class Manufacturer(models.Model):
try:
return cls.objects.get(slug=slug), False
except cls.DoesNotExist:
# Check historical slugs
from history_tracking.models import HistoricalSlug
# Check pghistory first
history_model = cls.get_history_model()
history_entry = (
history_model.objects.filter(slug=slug)
.order_by('-pgh_created_at')
.first()
)
if history_entry:
return cls.objects.get(id=history_entry.pgh_obj_id), True
# Check manual slug history as fallback
try:
historical = HistoricalSlug.objects.get(
content_type__model='manufacturer',
@@ -88,43 +109,3 @@ class Manufacturer(models.Model):
return cls.objects.get(pk=historical.object_id), True
except (HistoricalSlug.DoesNotExist, cls.DoesNotExist):
raise cls.DoesNotExist()
class Designer(models.Model):
name = models.CharField(max_length=255)
slug = models.SlugField(max_length=255, unique=True)
website = models.URLField(blank=True)
description = models.TextField(blank=True)
total_rides = models.IntegerField(default=0)
total_roller_coasters = models.IntegerField(default=0)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
objects: ClassVar[models.Manager['Designer']]
class Meta:
ordering = ['name']
def __str__(self) -> str:
return self.name
def save(self, *args, **kwargs) -> None:
if not self.slug:
self.slug = slugify(self.name)
super().save(*args, **kwargs)
@classmethod
def get_by_slug(cls, slug: str) -> Tuple['Designer', bool]:
"""Get designer by slug, checking historical slugs if needed"""
try:
return cls.objects.get(slug=slug), False
except cls.DoesNotExist:
# Check historical slugs
from history_tracking.models import HistoricalSlug
try:
historical = HistoricalSlug.objects.get(
content_type__model='designer',
slug=slug
)
return cls.objects.get(pk=historical.object_id), True
except (HistoricalSlug.DoesNotExist, cls.DoesNotExist):
raise cls.DoesNotExist()