from django.db import models from django.contrib.contenttypes.models import ContentType from django.contrib.contenttypes.fields import GenericForeignKey from django.contrib.auth import get_user_model from simple_history.models import HistoricalRecords from .mixins import HistoricalChangeMixin from typing import Any, Type, TypeVar, cast, Optional from django.db.models import QuerySet from django.core.exceptions import ValidationError from django.utils import timezone T = TypeVar('T', bound=models.Model) User = get_user_model() class HistoricalModel(models.Model): """Abstract base class for models with history tracking""" id = models.BigAutoField(primary_key=True) history: HistoricalRecords = HistoricalRecords( inherit=True, bases=(HistoricalChangeMixin,) ) class Meta: abstract = True @property def _history_model(self) -> Type[T]: """Get the history model class""" return cast(Type[T], self.history.model) # type: ignore def get_history(self) -> QuerySet: """Get all history records for this instance""" model = self._history_model return model.objects.filter(id=self.pk).order_by('-history_date') class HistoricalSlug(models.Model): """Track historical slugs for models""" content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE) object_id = models.PositiveIntegerField() content_object = GenericForeignKey('content_type', 'object_id') slug = models.SlugField(max_length=255) created_at = models.DateTimeField(auto_now_add=True) class Meta: unique_together = ('content_type', 'slug') indexes = [ models.Index(fields=['content_type', 'object_id']), models.Index(fields=['slug']), ] def __str__(self) -> str: return f"{self.content_type} - {self.object_id} - {self.slug}" class VersionBranch(models.Model): """Represents a version control branch for tracking parallel development""" name = models.CharField(max_length=255, unique=True) parent = models.ForeignKey('self', null=True, blank=True, on_delete=models.CASCADE, related_name='children') created_at = models.DateTimeField(auto_now_add=True) created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True) metadata = models.JSONField(default=dict, blank=True) is_active = models.BooleanField(default=True) class Meta: ordering = ['-created_at'] indexes = [ models.Index(fields=['name']), models.Index(fields=['parent']), models.Index(fields=['created_at']), ] def __str__(self) -> str: return f"{self.name} ({'active' if self.is_active else 'inactive'})" def clean(self) -> None: # Prevent circular references if self.parent and self.pk: branch = self.parent while branch: if branch.pk == self.pk: raise ValidationError("Circular branch reference detected") branch = branch.parent class VersionTag(models.Model): """Tags specific versions for reference (releases, milestones, etc)""" name = models.CharField(max_length=255, unique=True) branch = models.ForeignKey(VersionBranch, on_delete=models.CASCADE, related_name='tags') content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE) object_id = models.PositiveIntegerField() historical_instance = GenericForeignKey('content_type', 'object_id') created_at = models.DateTimeField(auto_now_add=True) created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True) metadata = models.JSONField(default=dict, blank=True) class Meta: ordering = ['-created_at'] indexes = [ models.Index(fields=['name']), models.Index(fields=['branch']), models.Index(fields=['created_at']), models.Index(fields=['content_type', 'object_id']), ] def __str__(self) -> str: return f"{self.name} ({self.branch.name})" class ChangeSet(models.Model): """Groups related changes together for atomic version control operations""" branch = models.ForeignKey(VersionBranch, on_delete=models.CASCADE, related_name='changesets') created_at = models.DateTimeField(auto_now_add=True) created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True) description = models.TextField(blank=True) metadata = models.JSONField(default=dict, blank=True) dependencies = models.JSONField(default=dict, blank=True) status = models.CharField( max_length=20, choices=[ ('pending', 'Pending'), ('applied', 'Applied'), ('failed', 'Failed'), ('reverted', 'Reverted') ], default='pending' ) # Instead of directly relating to HistoricalRecord, use GenericForeignKey content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE) object_id = models.PositiveIntegerField() historical_instance = GenericForeignKey('content_type', 'object_id') class Meta: ordering = ['-created_at'] indexes = [ models.Index(fields=['branch']), models.Index(fields=['created_at']), models.Index(fields=['status']), models.Index(fields=['content_type', 'object_id']), ] def __str__(self) -> str: return f"ChangeSet {self.pk} ({self.branch.name} - {self.status})" def apply(self) -> None: """Apply the changeset to the target branch""" if self.status != 'pending': raise ValidationError(f"Cannot apply changeset with status: {self.status}") try: # Apply changes through the historical instance if self.historical_instance: instance = self.historical_instance.instance if instance: instance.save() self.status = 'applied' except Exception as e: self.status = 'failed' self.metadata['error'] = str(e) self.save() def revert(self) -> None: """Revert the changes in this changeset""" if self.status != 'applied': raise ValidationError(f"Cannot revert changeset with status: {self.status}") try: # Revert changes through the historical instance if self.historical_instance: instance = self.historical_instance.instance if instance: instance.save() self.status = 'reverted' except Exception as e: self.metadata['revert_error'] = str(e) self.save()