Source code for debusine.db.models.collections

# Copyright 2024 The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.

"""Data models for db collections."""

from datetime import datetime
from functools import cached_property
from typing import Any, TYPE_CHECKING, TypeAlias

from django.conf import settings
from django.contrib.postgres.constraints import ExclusionConstraint
from django.contrib.postgres.fields import RangeOperators
from django.db import models
from django.db.models import (
    CheckConstraint,
    F,
    Q,
    QuerySet,
    UniqueConstraint,
)
from django.urls import reverse

import jsonpath_rw

from debusine.artifacts.models import BareDataCategory, CollectionCategory
from debusine.db.constraints import JsonDataUniqueConstraint
from debusine.db.models.workspaces import Workspace

if TYPE_CHECKING:
    from django_stubs_ext.db.models import TypedModelMeta

    from debusine.db.models.artifacts import Artifact
    from debusine.db.models.auth import User
    from debusine.server.collections import CollectionManagerInterface
else:
    TypedModelMeta = object


class _CollectionRetainsArtifacts(models.TextChoices):
    """Choices for Collection.retains_artifacts."""

    NEVER = "never", "Never"
    WORKFLOW = "workflow", "While workflow is running"
    ALWAYS = "always", "Always"


[docs]class Collection(models.Model): """Model representing a collection.""" RetainsArtifacts: TypeAlias = _CollectionRetainsArtifacts name = models.CharField(max_length=255) category = models.CharField(max_length=255) full_history_retention_period = models.DurationField(null=True, blank=True) metadata_only_retention_period = models.DurationField(null=True, blank=True) workspace = models.ForeignKey(Workspace, on_delete=models.PROTECT) retains_artifacts = models.CharField( max_length=8, choices=RetainsArtifacts.choices, default=RetainsArtifacts.ALWAYS, ) child_artifacts = models.ManyToManyField( "db.Artifact", through="db.CollectionItem", through_fields=("parent_collection", "artifact"), related_name="parent_collections", ) child_collections = models.ManyToManyField( "self", through="db.CollectionItem", through_fields=("parent_collection", "collection"), related_name="parent_collections", symmetrical=False, ) data = models.JSONField(default=dict, blank=True) class Meta(TypedModelMeta): constraints = [ UniqueConstraint( fields=["name", "category", "workspace"], name="%(app_label)s_%(class)s_unique_name_category_workspace", ), CheckConstraint( check=~Q(name=""), name="%(app_label)s_%(class)s_name_not_empty" ), CheckConstraint( check=~Q(category=""), name="%(app_label)s_%(class)s_category_not_empty", ), ] @cached_property def manager(self) -> "CollectionManagerInterface": """Get collection manager for this collection category.""" # Local import to avoid circular dependency from debusine.server.collections import CollectionManagerInterface return CollectionManagerInterface.get_manager_for(self)
[docs] def get_absolute_url(self) -> str: """Return the canonical URL to display the item.""" return reverse( "workspaces:collections:detail", kwargs={ "wname": self.workspace.name, "ccat": self.category, "cname": self.name, }, )
def __str__(self) -> str: """Return id, name, category.""" return f"Id: {self.id} Name: {self.name} Category: {self.category}"
class CollectionItemManager(models.Manager["CollectionItem"]): """Manager for CollectionItem model.""" @staticmethod def create_from_bare_data( category: BareDataCategory, *, parent_collection: Collection, name: str, data: dict[str, Any], created_by_user: "User", ) -> "CollectionItem": """Create a CollectionItem from bare data.""" return CollectionItem.objects.create( parent_collection=parent_collection, name=name, child_type=CollectionItem.Types.BARE, category=category, data=data, created_by_user=created_by_user, ) @staticmethod def create_from_artifact( artifact: "Artifact", *, parent_collection: Collection, name: str, data: dict[str, Any], created_by_user: "User", ) -> "CollectionItem": """Create a CollectionItem from the artifact.""" return CollectionItem.objects.create( parent_collection=parent_collection, name=name, artifact=artifact, child_type=CollectionItem.Types.ARTIFACT, category=artifact.category, data=data, created_by_user=created_by_user, ) @staticmethod def create_from_collection( collection: Collection, *, parent_collection: Collection, name: str, data: dict[str, Any], created_by_user: "User", ) -> "CollectionItem": """Create a CollectionItem from the collection.""" return CollectionItem.objects.create( parent_collection=parent_collection, name=name, collection=collection, child_type=CollectionItem.Types.COLLECTION, category=collection.category, data=data, created_by_user=created_by_user, ) def drop_full_history(self, at: datetime): """ Drop artifacts from collections after full_history_retention_period. :param at: datetime to check if the artifacts are old enough. """ self.get_queryset().exclude(removed_at__isnull=True).filter( # https://github.com/typeddjango/django-stubs/issues/1548 removed_at__lt=( at - F("parent_collection__full_history_retention_period") # type: ignore[operator] # noqa: E501 ) ).update(artifact=None) def drop_metadata(self, at: datetime) -> None: """ Delete old collection items. After full_history_retention_period + metadata_only_retention_period. :param at: datetime to check if the collection item is old enough. """ self.get_queryset().exclude(removed_at__isnull=True).filter( # https://github.com/typeddjango/django-stubs/issues/1548 removed_at__lt=( at - F("parent_collection__full_history_retention_period") - F("parent_collection__metadata_only_retention_period") # type: ignore[operator] # noqa: E501 ) ).delete() class CollectionItemActiveManager(CollectionItemManager): """Manager for active collection items.""" def get_queryset(self) -> QuerySet["CollectionItem"]: """Return only active collection items.""" return super().get_queryset().filter(removed_at__isnull=True) class _CollectionItemTypes(models.TextChoices): """Choices for the CollectionItem.type.""" BARE = "b", "Bare" ARTIFACT = "a", "Artifact" COLLECTION = "c", "Collection"
[docs]class CollectionItem(models.Model): """CollectionItem model.""" objects = CollectionItemManager() active_objects = CollectionItemActiveManager() name = models.CharField(max_length=255) Types: TypeAlias = _CollectionItemTypes child_type = models.CharField(max_length=1, choices=Types.choices) # category duplicates the category of the artifact or collection of this # item, so when the underlying artifact or collection is deleted the # category is retained category = models.CharField(max_length=255) parent_collection = models.ForeignKey( Collection, on_delete=models.PROTECT, related_name="child_items", ) parent_category = models.CharField(max_length=255) collection = models.ForeignKey( Collection, on_delete=models.PROTECT, related_name="collection_items", null=True, ) artifact = models.ForeignKey( "Artifact", on_delete=models.PROTECT, related_name="collection_items", null=True, ) data = models.JSONField(default=dict, blank=True) created_at = models.DateTimeField(auto_now_add=True) created_by_user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="user_created_%(class)s", ) # created_by_workflow = models.ForeignKey(Workflow, # on_delete=models.PROTECT, null=True) removed_at = models.DateTimeField(blank=True, null=True) removed_by_user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.PROTECT, null=True, related_name="user_removed_%(class)s", ) # removed_by_workflow = models.ForeignKey(Workflow, # on_delete=models.PROTECT, null=True)
[docs] @staticmethod def expand_variables( variables: dict[str, str], artifact: "Artifact" ) -> dict[str, str]: """Expand JSONPath variables against an Artifact's data.""" jsonpaths = {} for name, path in variables.items(): if name.startswith("$"): try: jsonpaths[name[1:]] = jsonpath_rw.parse(path) except Exception as e: raise ValueError(e) expanded_variables = {} for name, jsonpath in jsonpaths.items(): matches = jsonpath.find(artifact.data) if len(matches) == 1: expanded_variables[name] = matches[0].value elif len(matches) > 1: raise ValueError( "Too many values expanding", variables[f"${name}"], artifact.data, ) else: raise KeyError(variables[f"${name}"], artifact.data) for name, value in variables.items(): if not name.startswith("$"): if name in expanded_variables: raise ValueError( f"Cannot set both '${name}' and '{name}' variables" ) expanded_variables[name] = value return expanded_variables
[docs] @staticmethod def expand_name( name_template: str, expanded_variables: dict[str, str] ) -> str: """Format item name following item_template.""" return name_template.format(**expanded_variables)
[docs] def get_absolute_url(self) -> str: """Return the canonical URL to display the item.""" return reverse( "workspaces:collections:item_detail", kwargs={ "wname": self.parent_collection.workspace.name, "ccat": self.parent_collection.category, "cname": self.parent_collection.name, "iid": self.pk, "iname": self.name, }, )
def __str__(self) -> str: """Return id, name, collection_id, child_type.""" item_info = ( f" Artifact id: {self.artifact.id}" if self.artifact else ( f" Collection id: {self.collection.id}" if self.collection else "" ) ) return ( f"Id: {self.id} Name: {self.name} " f"Parent collection id: {self.parent_collection_id} " f"Child type: {self.child_type}" f"{item_info}" ) class Meta(TypedModelMeta): constraints = [ JsonDataUniqueConstraint( fields=[ "category", "data->>'codename'", "data->>'architecture'", "data->>'variant'", "data->>'backend'", "parent_collection", ], condition=Q( parent_category=CollectionCategory.ENVIRONMENTS, removed_at__isnull=True, ), nulls_distinct=False, name="%(app_label)s_%(class)s_unique_debian_environments", ), UniqueConstraint( fields=["name", "parent_collection"], condition=Q(removed_at__isnull=True), name="%(app_label)s_%(class)s_unique_active_name", ), # Prevent direct way to add a collection to itself. # It is still possible to add loops of collections. The Manager # should avoid it CheckConstraint( check=~Q(collection=F("parent_collection")), name="%(app_label)s_%(class)s_distinct_parent_collection", ), CheckConstraint( name="%(app_label)s_%(class)s_childtype_removedat_consistent", check=( Q( child_type=_CollectionItemTypes.BARE, collection__isnull=True, artifact__isnull=True, ) | ( Q( child_type=_CollectionItemTypes.ARTIFACT, collection__isnull=True, ) & ( Q(artifact__isnull=False) | Q(removed_at__isnull=False) ) ) | ( Q( child_type=_CollectionItemTypes.COLLECTION, artifact__isnull=True, ) & ( Q(collection__isnull=False) | Q(removed_at__isnull=False) ) ) ), ), ]
[docs] def save(self, *args: Any, **kwargs: Any) -> None: """Populate `parent_category` on save.""" if not self.parent_category: self.parent_category = self.parent_collection.category super().save(*args, **kwargs)
[docs]class CollectionItemMatchConstraint(models.Model): """ Enforce matching-value constraints on collection items. All instances of this model with the same :py:attr:`collection`, :py:attr:`constraint_name`, and :py:attr:`key` must have the same :py:attr:`value`. """ objects = models.Manager["CollectionItemMatchConstraint"]() collection = models.ForeignKey( Collection, on_delete=models.CASCADE, related_name="item_match_constraints", ) # This is deliberately a bare ID, not a foreign key: some constraints # take into account even items that no longer exist but were in a # collection in the past. collection_item_id = models.BigIntegerField() constraint_name = models.CharField(max_length=255) key = models.TextField() value = models.TextField() class Meta(TypedModelMeta): constraints = [ ExclusionConstraint( name="%(app_label)s_%(class)s_match_value", expressions=( (F("collection"), RangeOperators.EQUAL), (F("constraint_name"), RangeOperators.EQUAL), (F("key"), RangeOperators.EQUAL), (F("value"), RangeOperators.NOT_EQUAL), ), ) ] indexes = [ models.Index( name="%(app_label)s_cimc_collection_item_idx", fields=["collection_item_id"], ) ]