# Copyright © 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."""
import datetime as dt
import enum
import re
from collections.abc import Generator
from functools import cached_property
from typing import (
Any,
ClassVar,
Optional,
Self,
TYPE_CHECKING,
TypeAlias,
assert_never,
cast,
override,
)
import jsonpath_rw
from django.conf import settings
from django.contrib.postgres.constraints import ExclusionConstraint
from django.contrib.postgres.fields import RangeOperators
from django.core.exceptions import ImproperlyConfigured, ValidationError
from django.db import models, transaction
from django.db.models import (
CheckConstraint,
Count,
Deferrable,
Exists,
F,
JSONField,
Max,
OuterRef,
Q,
QuerySet,
UniqueConstraint,
Value,
)
from django.db.models.fields.json import KT
from django.db.models.functions import Coalesce
from django.urls import reverse
from django.utils import timezone
from debusine.artifacts.models import (
ArtifactCategory,
BareDataCategory,
BaseArtifactDataModel,
CollectionCategory,
DebusinePromise,
SINGLETON_COLLECTION_CATEGORIES,
)
from debusine.db import COLLATION_CODEPOINT, COLLATION_PRESENTATION
from debusine.db.context import context
from debusine.db.models.permissions import (
Allow,
FastCheckResult,
Role,
RoleBase,
Roles,
fast_check_common,
permission_check,
permission_filter,
)
from debusine.db.models.workspaces import Workspace
from debusine.db.permissioncontext import PermissionContext
if TYPE_CHECKING:
from django.http import HttpRequest
from django_stubs_ext.db.models import TypedModelMeta
from debusine.db.models.artifacts import Artifact
from debusine.db.models.auth import Token, User
from debusine.db.models.work_requests import WorkRequest
from debusine.server.collections import CollectionManagerInterface
from debusine.web.views.ui.collection_relations import CollectionRelationUI
from debusine.web.views.ui.collections import CollectionItemUI, CollectionUI
else:
TypedModelMeta = object
#: Regexp matching the structure of collection names
collection_name_regex = re.compile(r"^[A-Za-z][A-Za-z0-9+._-]*$")
def is_valid_collection_name(value: str) -> bool:
"""Check if value is a valid scope name."""
return value == "_" or bool(collection_name_regex.match(value))
def validate_collection_name(value: str) -> None:
"""Validate collection names."""
if not is_valid_collection_name(value):
raise ValidationError(
"%(value)r is not a valid collection name", params={"value": value}
)
class _CollectionRetainsArtifacts(models.TextChoices):
"""Choices for Collection.retains_artifacts."""
NEVER = "never", "Never"
WORKFLOW = "workflow", "While workflow is running"
ALWAYS = "always", "Always"
class CollectionRoleBase(RoleBase):
"""Collection role implementation."""
implied_by_workspace_roles: frozenset[Workspace.Roles]
implied_by_collection_roles: frozenset["CollectionRoles"]
@override
def _setup(self) -> None:
"""Set up implications for a newly constructed role."""
implied_by_workspace_roles: set[Workspace.Roles] = set()
implied_by_collection_roles: set[CollectionRoles] = {
cast(CollectionRoles, self)
}
for i in self.implied_by:
match i:
case Workspace.Roles():
implied_by_workspace_roles |= i.implied_by_workspace_roles
case Role():
# Resolve a role passed during class definition into its
# enum instance
role = self.__class__(i.value)
implied_by_workspace_roles |= (
role.implied_by_workspace_roles
)
implied_by_collection_roles |= (
role.implied_by_collection_roles
)
case _:
raise ImproperlyConfigured(
f"Collection roles do not support implications by {i!r}"
)
self.implied_by_workspace_roles = frozenset(implied_by_workspace_roles)
self.implied_by_collection_roles = frozenset(
implied_by_collection_roles
)
@classmethod
def q(cls, pc: PermissionContext, *roles: "CollectionRoleBase") -> Q:
"""Return a Q expression to select collections with this role."""
workspace_roles = set().union(
*(role.implied_by_workspace_roles for role in roles)
)
q = Q(
workspace__in=Workspace.objects.filter(
Workspace.Roles.q(pc, *workspace_roles)
)
)
if pc.user.is_authenticated:
q |= Q(
roles__group__in=pc.groups(),
roles__role__in=set().union(
*(role.implied_by_collection_roles for role in roles)
),
)
q = Q(workspace__in=Workspace.objects.can_display(pc)) & q
explicitly_granted = {
grant.resource.pk
for grant in pc.extra_resource_grants
if (
isinstance(grant.resource, Collection)
and any(grant.role.implies(role) for role in roles)
)
}
q |= Q(pk__in=explicitly_granted)
return q
@classmethod
def fast_check(
cls,
pc: PermissionContext,
resource: "Collection",
*roles: "Collection.Roles",
) -> FastCheckResult:
"""
Check if the context can satisfy one of the given roles.
:param pc: permission context to check
:param resource: resource to check
:param roles: roles that would satisfy the check
:returns: YES if at least one of the given roles is satisfied without
needing DB lookups. NO if it can be decided that none of the given
roles is satisfied without needing DB lookup. UNDECIDED if DB lookup
is needed to decide.
"""
if (
res := fast_check_common(pc, resource, *roles)
) != FastCheckResult.UNDECIDED:
return res
if not context.pc_is(pc):
return FastCheckResult.UNDECIDED
if resource.workspace != context.workspace:
return FastCheckResult.UNDECIDED
# Allowed as implied by workspace roles the user has
workspace_roles = frozenset.union(
*(role.implied_by_workspace_roles for role in roles)
)
if (
workspace_roles
and Workspace.Roles.fast_check(
pc, resource.workspace, *workspace_roles
)
== FastCheckResult.YES
):
return FastCheckResult.YES
return FastCheckResult.UNDECIDED
def implies(self, role: "CollectionRoles") -> bool:
"""Check if this role implies the given one."""
return (
self.implied_by_workspace_roles <= role.implied_by_workspace_roles
and self.implied_by_collection_roles
<= role.implied_by_collection_roles
)
class CollectionRoles(Roles, CollectionRoleBase, enum.ReprEnum):
"""Available roles for a Collection."""
OWNER = Role("owner", implied_by=[Workspace.Roles.OWNER])
VIEWER = Role("viewer", implied_by=[OWNER, Workspace.Roles.VIEWER])
CollectionRoles.setup()
class CollectionQuerySet[A](QuerySet["Collection", A]):
"""Custom QuerySet for Collection."""
def in_current_scope(self) -> Self:
"""Filter to collections in the current scope."""
from debusine.db.context import context
return self.filter(workspace__scope=context.require_scope())
def in_current_workspace(self) -> Self:
"""Filter to collections in the current workspace."""
from debusine.db.context import context
return self.filter(workspace=context.require_workspace())
def with_role(self, pc: PermissionContext, role: CollectionRoles) -> Self:
"""Keep only resources where the user has the given role."""
return self.filter(
Exists(
self.model.objects.filter(
CollectionRoles.q(pc, role), pk=OuterRef("pk")
)
)
)
@permission_filter(
work_request=Allow.PASS,
anonymous=Allow.PASS,
explicit_grants_role=CollectionRoles.VIEWER,
)
def can_display(self, pc: PermissionContext) -> Self:
"""Keep only Collections that can be displayed."""
return self.with_role(pc, CollectionRoles.VIEWER)
@permission_filter()
def can_configure(self, pc: PermissionContext) -> Self:
"""Keep only Collections that can be configured."""
return self.with_role(pc, CollectionRoles.OWNER)
@permission_filter()
def can_create_collection_relation(self, pc: PermissionContext) -> Self:
"""Keep only Collections that can have a collection relation added."""
return self.with_role(pc, CollectionRoles.OWNER)
# TODO: Ideally we'd lock down work request access to edit collection
# contents a bit more, but that requires more infrastructure similar to
# BaseTask.get_input_artifacts_ids.
@permission_filter(
work_request=Allow.PASS, explicit_grants_role=CollectionRoles.OWNER
)
def can_edit_contents(self, pc: PermissionContext) -> Self:
"""Keep only Collections whose contents can be edited."""
return self.with_role(pc, CollectionRoles.OWNER)
@permission_filter()
def can_delete(self, pc: PermissionContext) -> Self:
"""Keep only Collections that can be deleted."""
return self.with_role(pc, CollectionRoles.OWNER).exclude(
category=CollectionCategory.WORKFLOW_INTERNAL
)
@permission_filter()
def can_assign_roles(self, pc: PermissionContext) -> Self:
"""Keep only Collections whose roles can be assigned to groups."""
return self.with_role(pc, CollectionRoles.OWNER).exclude(
category=CollectionCategory.WORKFLOW_INTERNAL
)
def exported_suites(self) -> "CollectionQuerySet[Any]":
"""Filter to exported ``debian:suite`` collections."""
return (
self.filter(category=CollectionCategory.SUITE)
.annotate(
exported=Coalesce(
"data__exported", Value("true"), output_field=JSONField()
)
)
.exclude(exported=False)
)
class CollectionManager(models.Manager["Collection"]):
"""Manager for Collection model."""
def get_roles_model(self) -> type["CollectionRole"]:
"""Get the model used for role assignment."""
return CollectionRole
@override
def get_queryset(self) -> CollectionQuerySet[Any]:
"""Use the custom QuerySet."""
return CollectionQuerySet(self.model, using=self._db)
@staticmethod
def get_or_create_singleton(
category: CollectionCategory,
workspace: Workspace,
*,
data: dict[str, Any] | None = None,
) -> "tuple[Collection, bool]":
"""Create a singleton collection."""
if category not in SINGLETON_COLLECTION_CATEGORIES:
raise ValueError(
f"'{category}' is not a singleton collection category"
)
return Collection.objects.get_or_create(
name="_", category=category, workspace=workspace, data=data or {}
)
[docs]
class Collection(models.Model):
"""Model representing a collection."""
Roles: TypeAlias = CollectionRoles
RetainsArtifacts: TypeAlias = _CollectionRetainsArtifacts
name = models.CharField(
max_length=255,
validators=[validate_collection_name],
db_collation=COLLATION_PRESENTATION,
)
category = models.CharField(
max_length=255,
# TODO: db_collation=COLLATION_CODEPOINT,
)
full_history_retention_period = models.DurationField(
null=True,
blank=True,
help_text="Period after removing an item from the collection"
" during which the related artifact is not allowed to expire",
)
metadata_only_retention_period = models.DurationField(
null=True,
blank=True,
help_text="Period after removing an item from the collection during"
" which its metadata is kept in the history of the collection",
)
workspace = models.ForeignKey(
Workspace, on_delete=models.PROTECT, related_name="collections"
)
retains_artifacts = models.CharField(
max_length=8,
choices=RetainsArtifacts.choices,
default=RetainsArtifacts.ALWAYS,
# TODO: db_collation=COLLATION_CODEPOINT,
help_text="Controls if artifacts in this collection should be retained"
" rather than being allowed to expire",
)
data = models.JSONField(default=dict, blank=True)
objects = CollectionManager.from_queryset(CollectionQuerySet)()
class Meta(TypedModelMeta):
base_manager_name = "objects"
constraints = [
UniqueConstraint(
fields=["name", "category", "workspace"],
name="%(app_label)s_%(class)s_unique_name_category_workspace",
),
CheckConstraint(
condition=~Q(name=""),
name="%(app_label)s_%(class)s_name_not_empty",
),
CheckConstraint(
condition=(
(
~Q(category__in=sorted(SINGLETON_COLLECTION_CATEGORIES))
& ~Q(name__startswith="_")
)
| Q(
category__in=sorted(SINGLETON_COLLECTION_CATEGORIES),
name="_",
)
),
name="%(app_label)s_%(class)s_name_not_reserved",
),
CheckConstraint(
condition=~Q(category=""),
name="%(app_label)s_%(class)s_category_not_empty",
),
]
@override
def __str__(self) -> str:
"""Return name@category."""
# Stringify using a valid lookup syntax
return f"{self.name}@{self.category}"
@override
def __repr__(self) -> str:
"""Return representation of the collection."""
return f"<Collection: {self.name}@{self.category} ({self.pk})>"
[docs]
def get_absolute_url(self) -> str:
"""Return the canonical URL to display the collection."""
from debusine.server.scopes import urlconf_scope
with urlconf_scope(self.workspace.scope.name):
return reverse(
"workspaces:collections:detail",
kwargs={
"wname": self.workspace.name,
"ccat": self.category,
"cname": self.name,
},
)
[docs]
def get_absolute_url_search(self) -> str:
"""Return the canonical URL to search inside the collection."""
from debusine.server.scopes import urlconf_scope
with urlconf_scope(self.workspace.scope.name):
return reverse(
"workspaces:collections:search",
kwargs={
"wname": self.workspace.name,
"ccat": self.category,
"cname": self.name,
},
)
[docs]
@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 ui(self, request: "HttpRequest") -> "CollectionUI":
"""Return a UI helper for this instance."""
from debusine.web.views.ui.collections import CollectionUI
return CollectionUI.get(request, self)
[docs]
def grant_role(
self, token: "Token", role: CollectionRoles, *, created_by: "User"
) -> "CollectionTokenGrant":
"""Grant a role on this collection to a token."""
assert token.explicit_grants_only
grant, _ = CollectionTokenGrant.objects.get_or_create(
token=token, collection=self, role=role, created_by=created_by
)
return grant
[docs]
def has_role(self, pc: PermissionContext, role: CollectionRoles) -> bool:
"""Check if the user has the given role on this Collection."""
match Collection.Roles.fast_check(pc, self, role):
case FastCheckResult.YES:
return True
case FastCheckResult.NO:
return False
case FastCheckResult.UNDECIDED:
return (
Collection.objects.with_role(pc, role)
.filter(pk=self.pk)
.exists()
)
case _ as unreachable:
assert_never(unreachable)
[docs]
@permission_check(
"{user} cannot display collection {resource}",
work_request=Allow.PASS,
anonymous=Allow.PASS,
explicit_grants_role=CollectionRoles.VIEWER,
)
def can_display(self, pc: PermissionContext) -> bool:
"""Check if the collection can be displayed."""
return self.has_role(pc, CollectionRoles.VIEWER)
# TODO: Ideally we'd lock down work request access to edit collection
# contents a bit more, but that requires more infrastructure similar to
# BaseTask.get_input_artifacts_ids.
[docs]
@permission_check(
"{user} cannot edit contents of collection {resource}",
work_request=Allow.PASS,
explicit_grants_role=CollectionRoles.OWNER,
)
def can_edit_contents(self, pc: PermissionContext) -> bool:
"""Check if the collection's contents can be edited."""
return self.has_role(pc, CollectionRoles.OWNER)
[docs]
@permission_check(
"{user} cannot delete collection {resource}",
)
def can_delete(self, pc: PermissionContext) -> bool:
"""Check if the collection can be deleted."""
return (
self.category != CollectionCategory.WORKFLOW_INTERNAL
and self.has_role(pc, CollectionRoles.OWNER)
)
[docs]
@permission_check(
"{user} cannot create collection relation in {resource}",
)
def can_create_collection_relation(self, pc: PermissionContext) -> bool:
"""Check if a relation can be created from this collection."""
return self.has_role(pc, CollectionRoles.OWNER)
[docs]
@permission_check(
"{user} cannot assign roles to {resource}",
)
def can_assign_roles(self, pc: PermissionContext) -> bool:
"""Check if collection roles can be assigned to groups."""
if self.category == CollectionCategory.WORKFLOW_INTERNAL:
return False
return self.has_role(pc, CollectionRoles.OWNER)
[docs]
def get_item_stats(
self,
) -> Generator[dict[str, Any]]:
"""Return a dict of item counts for each item type."""
for rec in (
self.child_items.values(
"child_type", "category", active=Q(removed_at__isnull=True)
)
.annotate(count=Count("id"))
.order_by("child_type", "category", "active")
):
child_type = CollectionItem.Types(rec["child_type"])
yield {
"type": child_type,
"category": rec["category"],
"active": rec["active"],
"count": rec["count"],
}
[docs]
def delete_collection(self) -> None:
"""
Delete this collection, and all its CollectionItem entries.
Does not delete the associated Collections or Artifacts.
"""
self.child_items.all().delete()
self.delete()
class CollectionRole(models.Model):
"""Role assignment for collection."""
resource = models.ForeignKey(
Collection,
on_delete=models.CASCADE,
related_name="roles",
)
group = models.ForeignKey(
"Group",
on_delete=models.CASCADE,
related_name="collection_roles",
)
role = models.CharField(
max_length=16,
choices=CollectionRoles.choices,
# TODO: db_collation=COLLATION_CODEPOINT,
)
class Meta(TypedModelMeta):
constraints = [
UniqueConstraint(
fields=["resource", "group", "role"],
name="%(app_label)s_%(class)s_unique_resource_group_role",
),
]
@override
def __str__(self) -> str:
"""Return a description of the role assignment."""
return f"{self.group}─{self.role}⟶{self.resource}"
class CollectionItemQuerySet(QuerySet["CollectionItem"]):
"""Custom QuerySet for CollectionItem."""
def active(self) -> "CollectionItemQuerySet":
"""Filter to only active collection items."""
return super().filter(removed_at__isnull=True)
def active_at(self, at: dt.datetime) -> "CollectionItemQuerySet":
"""Filter to collection items that were active at a given time."""
return self.filter(
Q(created_at__lte=at),
Q(removed_at__isnull=True) | Q(removed_at__gt=at),
)
def artifacts_in_collection(
self, parent_collection: Collection, category: ArtifactCategory
) -> "CollectionItemQuerySet":
"""
Filter to artifacts in a given parent collection.
This is tuned to use the indexes on :py:class:`CollectionItem`.
"""
return self.filter(
parent_collection=parent_collection,
# Technically redundant with parent_collection, but clues
# PostgreSQL into using the correct index.
parent_category=parent_collection.category,
child_type=CollectionItem.Types.ARTIFACT,
category=category,
)
# TODO: Ideally we'd lock down work request access to display collection
# items a bit more, but that requires more infrastructure similar to
# BaseTask.get_input_artifacts_ids.
@permission_filter(work_request=Allow.PASS)
def can_display(self, pc: PermissionContext) -> Self:
"""Keep only collection items that can be displayed."""
return self.filter(
parent_collection__in=Collection.objects.can_display(pc)
)
# TODO: Ideally we'd lock down work request access to delete collection
# items a bit more, but that requires more infrastructure similar to
# BaseTask.get_input_artifacts_ids.
@permission_filter(work_request=Allow.PASS)
def can_delete(self, pc: PermissionContext) -> Self:
"""Keep only collection items that can be deleted."""
return self.filter(
parent_collection__in=Collection.objects.can_edit_contents(pc)
)
class CollectionItemManager(models.Manager["CollectionItem"]):
"""Manager for CollectionItem model."""
@staticmethod
def create_from_bare_data(
category: BareDataCategory,
*,
parent_collection: Collection,
name: str,
data: BaseArtifactDataModel | dict[str, Any],
created_at: dt.datetime | None = None,
created_by_user: "User",
created_by_workflow: Optional["WorkRequest"] = None,
replaced_by: Optional["CollectionItem"] = None,
) -> "CollectionItem":
"""Create a CollectionItem from bare data."""
if isinstance(data, BaseArtifactDataModel):
data = data.model_dump(mode="json", exclude_unset=True)
match category:
case BareDataCategory.PROMISE:
# Raise ValueError if data is not valid
DebusinePromise(**data)
kwargs: dict[str, Any] = {}
if created_at is not None:
kwargs["created_at"] = created_at
if replaced_by is not None:
kwargs["removed_at"] = replaced_by.created_at
kwargs["removed_by_user"] = replaced_by.created_by_user
kwargs["removed_by_workflow"] = replaced_by.created_by_workflow
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,
created_by_workflow=created_by_workflow,
**kwargs,
)
@staticmethod
def create_from_artifact(
artifact: "Artifact",
*,
parent_collection: Collection,
name: str,
data: dict[str, Any],
created_at: dt.datetime | None = None,
created_by_user: "User",
created_by_workflow: Optional["WorkRequest"] = None,
replaced_by: Optional["CollectionItem"] = None,
) -> "CollectionItem":
"""Create a CollectionItem from the artifact."""
kwargs: dict[str, Any] = {}
if created_at is not None:
kwargs["created_at"] = created_at
if replaced_by is not None:
kwargs["removed_at"] = replaced_by.created_at
kwargs["removed_by_user"] = replaced_by.created_by_user
kwargs["removed_by_workflow"] = replaced_by.created_by_workflow
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,
created_by_workflow=created_by_workflow,
**kwargs,
)
@staticmethod
def create_from_collection(
collection: Collection,
*,
parent_collection: Collection,
name: str,
data: dict[str, Any],
created_at: dt.datetime | None = None,
created_by_user: "User",
created_by_workflow: Optional["WorkRequest"] = None,
replaced_by: Optional["CollectionItem"] = None,
) -> "CollectionItem":
"""Create a CollectionItem from the collection."""
kwargs: dict[str, Any] = {}
if created_at is not None:
kwargs["created_at"] = created_at
if replaced_by is not None:
kwargs["removed_at"] = replaced_by.created_at
kwargs["removed_by_user"] = replaced_by.created_by_user
kwargs["removed_by_workflow"] = replaced_by.created_by_workflow
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,
created_by_workflow=created_by_workflow,
**kwargs,
)
def drop_full_history(self, at: dt.datetime) -> None:
"""
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).exclude(
artifact__isnull=True
).filter(
removed_at__lt=(
at - F("parent_collection__full_history_retention_period")
)
).update(artifact=None)
def drop_metadata(self, at: dt.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(
removed_at__lt=(
at
- F("parent_collection__full_history_retention_period")
- F("parent_collection__metadata_only_retention_period")
)
).delete()
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."""
name = models.CharField(
max_length=255,
db_collation=COLLATION_CODEPOINT,
)
Types: TypeAlias = _CollectionItemTypes
child_type = models.CharField(
max_length=1,
choices=Types.choices,
# TODO: db_collation=COLLATION_CODEPOINT,
)
# 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,
# TODO: db_collation=COLLATION_CODEPOINT,
)
parent_collection = models.ForeignKey(
Collection,
on_delete=models.PROTECT,
related_name="child_items",
)
parent_category = models.CharField(
max_length=255,
# TODO: db_collation=COLLATION_CODEPOINT,
)
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(default=timezone.now)
created_by_user = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.PROTECT,
related_name="user_created_%(class)s",
)
created_by_workflow = models.ForeignKey(
"WorkRequest",
on_delete=models.SET_NULL,
null=True,
related_name="workflow_created_%(class)s",
)
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(
"WorkRequest",
on_delete=models.SET_NULL,
null=True,
related_name="workflow_removed_%(class)s",
)
objects = CollectionItemManager.from_queryset(CollectionItemQuerySet)()
class Meta(TypedModelMeta):
base_manager_name = "objects"
constraints = [
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(
condition=~Q(collection=F("parent_collection")),
name="%(app_label)s_%(class)s_distinct_parent_collection",
),
CheckConstraint(
name="%(app_label)s_%(class)s_childtype_removedat_consistent",
condition=(
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)
)
)
),
),
]
indexes = [
models.Index("name", name="%(app_label)s_ci_name_idx"),
models.Index(
F("parent_collection"),
KT("data__package"),
KT("data__version"),
name="%(app_label)s_ci_suite_source_idx",
condition=Q(
parent_category=CollectionCategory.SUITE,
child_type=_CollectionItemTypes.ARTIFACT,
category=ArtifactCategory.SOURCE_PACKAGE,
),
),
models.Index(
F("parent_collection"),
KT("data__srcpkg_name"),
KT("data__srcpkg_version"),
name="%(app_label)s_ci_suite_binary_source_idx",
condition=Q(
parent_category=CollectionCategory.SUITE,
child_type=_CollectionItemTypes.ARTIFACT,
category=ArtifactCategory.BINARY_PACKAGE,
),
),
models.Index(
F("parent_collection"),
KT("data__path"),
name="%(app_label)s_ci_suite_index_path_idx",
condition=Q(
parent_category=CollectionCategory.SUITE,
child_type=_CollectionItemTypes.ARTIFACT,
category=ArtifactCategory.REPOSITORY_INDEX,
),
),
models.Index(
F("parent_collection"),
KT("data__build_id"),
name="%(app_label)s_ci_suite_dbgsym_id_idx",
condition=Q(
parent_category=CollectionCategory.SUITE,
child_type=_CollectionItemTypes.ARTIFACT,
category=ArtifactCategory.DEBUG_SYMBOLS,
),
),
models.Index(
F("parent_collection"),
KT("data__path"),
name="%(app_label)s_ci_archive_index_path_idx",
condition=Q(
parent_category=CollectionCategory.ARCHIVE,
child_type=_CollectionItemTypes.ARTIFACT,
category=ArtifactCategory.REPOSITORY_INDEX,
),
),
# Optimize generating suite indexes.
models.Index(
F("parent_collection"),
KT("data__component"),
F("created_at"),
F("removed_at"),
name="%(app_label)s_ci_suite_all_sources_idx",
condition=Q(
parent_category=CollectionCategory.SUITE,
child_type=_CollectionItemTypes.ARTIFACT,
category=ArtifactCategory.SOURCE_PACKAGE,
),
),
models.Index(
F("parent_collection"),
KT("data__component"),
KT("data__architecture"),
F("created_at"),
F("removed_at"),
name="%(app_label)s_ci_suite_all_binaries_idx",
condition=Q(
parent_category=CollectionCategory.SUITE,
child_type=_CollectionItemTypes.ARTIFACT,
category=ArtifactCategory.BINARY_PACKAGE,
),
),
# Optimize latest:TASKNAME_PACKAGE_ARCHITECTURE lookups in
# debian:qa-results collections.
models.Index(
F("parent_collection"),
KT("data__task_name"),
KT("data__package"),
KT("data__architecture"),
name="%(app_label)s_ci_qa_results_latest_idx",
condition=Q(
parent_category=CollectionCategory.QA_RESULTS,
removed_at__isnull=True,
),
),
]
@override
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}"
)
[docs]
@override
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]
def get_absolute_url(self) -> str:
"""Return the canonical URL to display the item."""
from debusine.server.scopes import urlconf_scope
with urlconf_scope(self.parent_collection.workspace.scope.name):
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,
},
)
[docs]
@staticmethod
def expand_variables(
variables: dict[str, str], reference_data: dict[Any, Any]
) -> dict[str, str]:
"""
Expand JSONPath variables against some data.
The data will normally come from an Artifact.
"""
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(reference_data)
if len(matches) == 1:
expanded_variables[name] = matches[0].value
elif len(matches) > 1:
raise ValueError(
"Too many values expanding",
variables[f"${name}"],
reference_data,
)
else:
raise KeyError(variables[f"${name}"], reference_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 ui(self, request: "HttpRequest") -> "CollectionItemUI":
"""Return a UI helper for this instance."""
from debusine.web.views.ui.collections import CollectionItemUI
return CollectionItemUI.get(request, self)
# TODO: Ideally we'd lock down work request access to display collection
# items a bit more, but that requires more infrastructure similar to
# BaseTask.get_input_artifacts_ids.
[docs]
@permission_check(
"{user} cannot display collection item {resource}",
work_request=Allow.PASS,
)
def can_display(self, pc: PermissionContext) -> bool:
"""Check if the collection item can be displayed."""
return self.parent_collection.can_display(pc)
# TODO: Ideally we'd lock down work request access to delete collection
# items a bit more, but that requires more infrastructure similar to
# BaseTask.get_input_artifacts_ids.
[docs]
@permission_check(
"{user} cannot delete collection item {resource}",
work_request=Allow.PASS,
)
def can_delete(self, pc: PermissionContext) -> bool:
"""Check if the collection item can be deleted."""
return self.parent_collection.can_edit_contents(pc)
[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: ClassVar[models.Manager["CollectionItemMatchConstraint"]] = (
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. Also note that the collection item's parent
# collection may not be the same as this model's collection field: some
# collections impose indirect constraints via child collections.
collection_item_id = models.BigIntegerField()
constraint_name = models.CharField(
max_length=255,
# TODO: db_collation=COLLATION_CODEPOINT,
)
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"],
)
]
class CollectionRelationQuerySet[A](QuerySet["CollectionRelation", A]):
"""Custom QuerySet for CollectionRelation."""
@permission_filter()
def can_delete(self, pc: PermissionContext) -> Self:
"""
Check if the collection relation can be deleted.
It can be deleted if the source collection can be configured.
"""
return self.filter(source__in=Collection.objects.can_configure(pc))
class CollectionRelationManager(models.Manager["CollectionRelation"]):
"""Manager for CollectionRelation model."""
@override
def get_queryset(self) -> CollectionRelationQuerySet[Any]:
return CollectionRelationQuerySet(self.model, using=self._db)
class _CollectionRelationTypes(models.TextChoices):
"""Types of relations."""
SUITE_FORKED_FROM = "suite-forked-from", "Forked from"
SUITE_BASED_ON = "suite-based-on", "Based on"
SUITE_REQUIRES = "suite-requires", "Requires"
SUITE_TARGETING = "suite-targeting", "Targeting"
SUITE_DEFAULT_QA_RESULTS = "suite-default-qa-results", "Default QA results"
[docs]
class CollectionRelation(models.Model):
"""Model relationships between collections."""
Types: TypeAlias = _CollectionRelationTypes
source = models.ForeignKey(
Collection, on_delete=models.CASCADE, related_name="relations"
)
source_category = models.CharField(
max_length=255,
# TODO: db_collation=COLLATION_CODEPOINT,
)
target = models.ForeignKey(
Collection, on_delete=models.CASCADE, related_name="targeted_by"
)
target_category = models.CharField(
max_length=255,
# TODO: db_collation=COLLATION_CODEPOINT,
)
type = models.CharField(
max_length=32,
choices=_CollectionRelationTypes.choices,
# TODO: db_collation=COLLATION_CODEPOINT,
)
position = models.PositiveIntegerField(null=True, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
created_by_user = models.ForeignKey(
"User",
related_name="created_collection_relations",
on_delete=models.PROTECT,
)
updated_at = models.DateTimeField(null=True, blank=True)
updated_by_user = models.ForeignKey(
"User",
related_name="updated_collection_relations",
on_delete=models.PROTECT,
null=True,
)
objects = CollectionRelationManager.from_queryset(
CollectionRelationQuerySet
)()
class Meta(TypedModelMeta):
base_manager_name = "objects"
constraints = [
# Enforce that `requires` relations have a position, others not
models.CheckConstraint(
name="%(app_label)s_%(class)s_position",
condition=(
Q(
type=_CollectionRelationTypes.SUITE_REQUIRES,
position__isnull=False,
)
| Q(
type__in=[
_CollectionRelationTypes.SUITE_FORKED_FROM,
_CollectionRelationTypes.SUITE_BASED_ON,
_CollectionRelationTypes.SUITE_TARGETING,
_CollectionRelationTypes.SUITE_DEFAULT_QA_RESULTS,
],
position__isnull=True,
)
),
),
# Enforce that `requires` relations have unique positions per source
models.UniqueConstraint(
name="%(app_label)s_%(class)s_suite_requires_no_duplicate_position",
fields=["source", "type", "position"],
# this only applies for SUITE_REQUIRES: the other ones have
# cardinality max 1 enforced. Not using condition so it can
# be deferred.
deferrable=Deferrable.DEFERRED,
),
models.UniqueConstraint(
name="%(app_label)s_%(class)s_cardinality_checks_max_1",
fields=["source", "type"],
condition=(
Q(type=_CollectionRelationTypes.SUITE_FORKED_FROM)
| Q(type=_CollectionRelationTypes.SUITE_BASED_ON)
| Q(type=_CollectionRelationTypes.SUITE_TARGETING)
| Q(type=_CollectionRelationTypes.SUITE_DEFAULT_QA_RESULTS)
),
),
# Enforce that `requires` relations do not have repeated targets
models.UniqueConstraint(
name="%(app_label)s_%(class)s_suite_requires_no_duplicate_target",
fields=["source", "type", "target"],
condition=Q(type=_CollectionRelationTypes.SUITE_REQUIRES),
),
models.CheckConstraint(
name="%(app_label)s_%(class)s_categories_correct",
condition=(
# SUITE_FORKED_FROM: both must be suites
Q(
type=_CollectionRelationTypes.SUITE_FORKED_FROM,
source_category=CollectionCategory.SUITE,
target_category=CollectionCategory.SUITE,
)
# SUITE_BASED_ON: both must be suites
| Q(
type=_CollectionRelationTypes.SUITE_BASED_ON,
source_category=CollectionCategory.SUITE,
target_category=CollectionCategory.SUITE,
)
# SUITE_REQUIRES: both must be suites
| Q(
type=_CollectionRelationTypes.SUITE_REQUIRES,
source_category=CollectionCategory.SUITE,
target_category=CollectionCategory.SUITE,
)
# SUITE_TARGETING: both must be suites
| Q(
type=_CollectionRelationTypes.SUITE_TARGETING,
source_category=CollectionCategory.SUITE,
target_category=CollectionCategory.SUITE,
)
# SUITE_DEFAULT_QA_RESULTS: collection must be a suite,
# target must be qa-results
| Q(
type=_CollectionRelationTypes.SUITE_DEFAULT_QA_RESULTS,
source_category=CollectionCategory.SUITE,
target_category=CollectionCategory.QA_RESULTS,
)
),
),
]
@override
def __str__(self) -> str:
"""Return string representation of the CollectionRelation."""
if self.position is not None:
position_str = f", position {self.position}"
else:
position_str = ""
return (
f"{self.source.name} → {self.target.name} "
f"({CollectionRelation.Types(self.type).label}{position_str})"
)
[docs]
@override
def save(
self, *args: Any, updated_by_user: "User | None" = None, **kwargs: Any
) -> None:
"""Populate denormalized categories and enforce update audit fields."""
if self.pk is not None:
# Updating an existing CollectionRelation: check that the User
# is provided
if updated_by_user is None:
raise ValueError(
"updated_by_user must be set "
"when updating a CollectionRelation"
)
if "update_fields" in kwargs:
# "Model.save.update_fields" is not used in Debusine
# at the moment. In order to avoid having to deal with
# "update_fields" not having "update_by_user": just check that
# is not used in this particular case
raise ValueError(
"update_fields is not supported when updating "
"CollectionRelation"
)
self.updated_by_user = updated_by_user
self.updated_at = timezone.now()
self.position = self._position_for_create(
source=self.source,
relation_type=_CollectionRelationTypes(self.type),
wanted_position=self.position,
)
self.source_category = self.source.category
self.target_category = self.target.category
with transaction.atomic():
if self.position is not None:
CollectionRelation.objects.filter(
source=self.source,
type=self.type,
position__gte=self.position,
).update(position=F("position") + 1)
super().save(*args, **kwargs)
[docs]
@permission_check(
"{user} cannot delete collection relation {resource}",
)
def can_delete(self, pc: PermissionContext) -> bool:
"""
Check if the collection relation can be deleted.
It can be deleted if the source collection can be configured.
"""
return self.source.can_configure(pc)
[docs]
@classmethod
def get_max_position(
cls, collection: Collection, relation_type: _CollectionRelationTypes
) -> int:
"""
Return the current maximum position for a given relation type.
Return 0 if no relation of this type exists.
"""
return (
cls.objects.filter(
source=collection,
type=relation_type,
).aggregate(max_position=Max("position"))["max_position"]
or 0
)
@classmethod
def _position_for_create(
cls,
*,
source: Collection,
relation_type: _CollectionRelationTypes,
wanted_position: int | None,
) -> int | None:
"""
Return the normalized position for a new relation.
Keep positions contiguous from 1 to the maximum position.
"""
if wanted_position is None:
return None
current_max_position = cls.get_max_position(
collection=source, relation_type=relation_type
)
if wanted_position < 1:
return 1
elif wanted_position > current_max_position + 1:
return current_max_position + 1
else:
return wanted_position
[docs]
@override
def delete(self, *args: Any, **kwargs: Any) -> tuple[int, dict[str, int]]:
"""Delete this relation and shift later positions."""
with transaction.atomic():
result = super().delete(*args, **kwargs)
if self.position is not None:
CollectionRelation.objects.filter(
source=self.source,
type=self.type,
position__gt=self.position,
).update(
position=F("position") - 1,
)
return result
[docs]
def ui(self, request: "HttpRequest") -> "CollectionRelationUI":
"""Return a UI helper for this instance."""
from debusine.web.views.ui.collection_relations import (
CollectionRelationUI,
)
return CollectionRelationUI.get(request, self)
class CollectionTokenGrantQuerySet[A](QuerySet["CollectionTokenGrant", A]):
"""Custom QuerySet for CollectionTokenGrant."""
def granted_resources(self, role: RoleBase) -> CollectionQuerySet[Any]:
"""Find collections granted by this queryset with this role."""
assert isinstance(role, CollectionRoles)
role_grants = self.filter(role=role)
return Collection.objects.filter(
# An explicit grant is enough.
Q(token_grants__in=role_grants)
# Granting access to an archive grants access to all its suites.
| (
Q(category=CollectionCategory.SUITE)
& Exists(
Collection.objects.filter(
category=CollectionCategory.ARCHIVE,
workspace=OuterRef("workspace"),
token_grants__in=role_grants,
)
)
)
)
class CollectionTokenGrantManager(models.Manager["CollectionTokenGrant"]):
"""Manager for the CollectionTokenGrant model."""
@override
def get_queryset(self) -> CollectionTokenGrantQuerySet[Any]:
return CollectionTokenGrantQuerySet(self.model, using=self._db)
[docs]
class CollectionTokenGrant(models.Model):
"""A grant of a specific role on a collection to a token."""
collection = models.ForeignKey(
Collection, on_delete=models.CASCADE, related_name="token_grants"
)
token = models.ForeignKey(
"db.Token", on_delete=models.CASCADE, related_name="collection_grants"
)
role = models.CharField(
choices=CollectionRoles.choices,
# TODO: db_collation=COLLATION_CODEPOINT,
)
created_at = models.DateTimeField(default=timezone.now)
created_by = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.PROTECT,
related_name="user_created_%(class)s",
)
objects = CollectionTokenGrantManager.from_queryset(
CollectionTokenGrantQuerySet
)()
class Meta(TypedModelMeta):
constraints = [
UniqueConstraint(
fields=["token", "collection", "role"],
name="%(app_label)s_%(class)s_unique",
)
]