# 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 artifacts."""
import datetime as dt
from functools import partial
from pathlib import Path
from typing import Any, Optional, Self, TYPE_CHECKING, TypedDict, cast, override
from django.conf import settings
from django.contrib.auth.models import AnonymousUser
from django.core.exceptions import PermissionDenied, ValidationError
from django.db import connection, models, transaction
from django.db.models import (
Count,
Exists,
F,
Max,
OuterRef,
Q,
QuerySet,
UniqueConstraint,
)
from django.db.models.functions import Coalesce
from django.urls import reverse
from django_cte import CTE, with_cte
from debusine.artifacts import LocalArtifact
from debusine.artifacts.models import ArtifactCategory, ArtifactData
from debusine.db import COLLATION_CODEPOINT, COLLATION_PRESENTATION
from debusine.db.models.files import File
from debusine.db.models.permissions import (
Allow,
permission_check,
permission_filter,
)
from debusine.db.models.workspaces import Workspace
from debusine.db.permissioncontext import PermissionContext
from debusine.utils.typing_utils import copy_signature_from
if TYPE_CHECKING:
from django.http import HttpRequest
from django_stubs_ext import WithAnnotations
from django_stubs_ext.db.models import TypedModelMeta
from debusine.db.models.work_requests import WorkRequest
from debusine.web.views.ui.artifacts import ArtifactUI, FileInArtifactUI
ArtifactWithComplete = WithAnnotations["Artifact", "CompleteDict"]
WithAnnotations # fake usage for vulture
else:
TypedModelMeta = object
ARTIFACT_CATEGORY_ICON_NAMES = {
ArtifactCategory.AUTOPKGTEST: "folder",
ArtifactCategory.BINARY_PACKAGE: "folder",
ArtifactCategory.BINARY_PACKAGES: "folder",
ArtifactCategory.BLHC: "folder",
ArtifactCategory.DEBDIFF: "folder",
ArtifactCategory.LINTIAN: "folder",
ArtifactCategory.PACKAGE_BUILD_LOG: "journal-text",
ArtifactCategory.SIGNING_INPUT: "folder",
ArtifactCategory.SIGNING_OUTPUT: "folder",
ArtifactCategory.SOURCE_PACKAGE: "folder",
ArtifactCategory.SYSTEM_IMAGE: "folder",
ArtifactCategory.SYSTEM_TARBALL: "folder",
ArtifactCategory.UPLOAD: "folder",
ArtifactCategory.WORK_REQUEST_DEBUG_LOGS: "folder",
}
ARTIFACT_CATEGORY_SHORT_NAMES = {
ArtifactCategory.AUTOPKGTEST: "autopkgtest",
ArtifactCategory.BINARY_PACKAGE: "binary package",
ArtifactCategory.BINARY_PACKAGES: "binary packages",
ArtifactCategory.BLHC: "blhc report",
ArtifactCategory.DEBDIFF: "debdiff report",
ArtifactCategory.LINTIAN: "lintian report",
ArtifactCategory.PACKAGE_BUILD_LOG: "build log",
ArtifactCategory.SIGNING_INPUT: "signing input",
ArtifactCategory.SIGNING_OUTPUT: "signing output",
ArtifactCategory.SOURCE_PACKAGE: "source package",
ArtifactCategory.SYSTEM_IMAGE: "system image",
ArtifactCategory.SYSTEM_TARBALL: "system tar",
ArtifactCategory.UPLOAD: "package upload",
ArtifactCategory.WORK_REQUEST_DEBUG_LOGS: "debug log",
}
class CompleteDict(TypedDict):
"""Additional fields returned by ArtifactQuerySet.annotate_complete."""
complete: bool
class ArtifactQuerySet[A](QuerySet["Artifact", A]):
"""Custom QuerySet for Artifact."""
def in_current_scope(self) -> Self:
"""Filter to artifacts 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 artifacts in the current workspace."""
from debusine.db.context import context
return self.filter(workspace=context.require_workspace())
def visible_to_current_workspace(self) -> Self:
"""Filter to artifacts that the current workspace inherits from."""
from debusine.db.context import context
visible = CTE(
Workspace.objects.can_display(context.pc),
name="visible",
)
chain = CTE.recursive(
context.require_workspace().make_inheritance_chain_cte(visible),
name="chain",
)
return self.filter(
workspace__in=with_cte(
visible,
chain,
select=chain.join(Workspace, pk=chain.col.chain_parent),
)
)
def work_request_can_display(self, work_request: "WorkRequest") -> Self:
"""Keep only artifacts that the work request can display."""
task = work_request.get_task()
return self.filter(
Q(pk__in=task.get_input_artifacts_ids())
| Q(created_by_work_request=work_request)
)
def work_request_can_change(self, work_request: "WorkRequest") -> Self:
"""Keep only artifacts that the work request can change."""
return self.filter(created_by_work_request=work_request)
@permission_filter(work_request=Allow.PASS, anonymous=Allow.PASS)
def can_display(self, pc: PermissionContext) -> Self:
"""Keep only Artifacts that can be displayed."""
# Delegate to workspace can_display check
qs = self.filter(workspace__in=Workspace.objects.can_display(pc))
if pc.work_request is not None:
qs = qs.work_request_can_display(pc.work_request)
return qs
@permission_filter(work_request=Allow.PASS)
def can_add_files(self, pc: PermissionContext) -> Self:
"""Keep only Artifacts where files can be added."""
qs = self.filter(
workspace__in=Workspace.objects.can_create_artifacts(pc)
)
if pc.work_request is not None:
qs = qs.work_request_can_change(pc.work_request)
return qs
@permission_filter(work_request=Allow.PASS)
def can_add_relation(self, pc: PermissionContext) -> Self:
"""Keep only Artifacts where relations can be added."""
qs = self.filter(
workspace__in=Workspace.objects.can_create_artifacts(pc)
)
if pc.work_request is not None:
qs = qs.work_request_can_change(pc.work_request)
return qs
def annotate_complete(
self,
) -> "ArtifactQuerySet[ArtifactWithComplete]":
"""Annotate artifacts with whether all their files are complete."""
return self.annotate(
complete=~Exists(
FileInArtifact.objects.filter(
artifact=OuterRef("pk"), complete=False
)
)
)
def not_expired(self, at: dt.datetime) -> Self:
"""
Return queryset with artifacts that have not expired.
:param at: datetime to check if the artifacts are not expired.
:return: artifacts that expire_at is None (do not expire) or
expire_at is after the given datetime.
"""
return self.annotate(
_effective_expiration_delay=Coalesce(
"expiration_delay",
"workspace__default_expiration_delay",
)
).filter(
Q(_effective_expiration_delay=dt.timedelta(0))
| Q(created_at__gt=at - F("_effective_expiration_delay"))
)
def expired(self, at: dt.datetime) -> Self:
"""
Return queryset with artifacts that have expired.
:param at: datetime to check if the artifacts are expired.
:return: artifacts that expire_at is before the given datetime.
"""
return (
self.annotate(
_effective_expiration_delay=Coalesce(
"expiration_delay",
"workspace__default_expiration_delay",
)
)
.exclude(_effective_expiration_delay=dt.timedelta(0))
.filter(created_at__lte=at - F("_effective_expiration_delay"))
)
def part_of_collection_with_retains_artifacts(self) -> Self:
"""
Return Artifacts that are in collections with retains_artifacts set.
:return: Artifacts that are part of a retains_artifacts collection.
"""
# Import here to prevent circular imports
from debusine.db.models.collections import Collection
from debusine.db.models.work_requests import WorkRequest
RetainsArtifacts = Collection.RetainsArtifacts
return self.filter(
Q(
collection_items__parent_collection__retains_artifacts=(
RetainsArtifacts.ALWAYS
)
)
| Q(
collection_items__parent_collection__retains_artifacts=(
RetainsArtifacts.WORKFLOW
),
collection_items__parent_collection__workflow__status__in={
WorkRequest.Statuses.PENDING,
WorkRequest.Statuses.RUNNING,
WorkRequest.Statuses.BLOCKED,
},
)
)
def retained_as_referenced(self) -> Self:
"""
Select artifacts to be kept as referenced by other artifacts.
The following artifact relations cause ``expired_artifact`` to be
kept:
* artifact=non_expired_artifact, target=expired_artifact, type=(any)
* artifact=expired_artifact, target=non_expired_artifact, type=EXTENDS
"""
def keep_cte(cte: CTE) -> QuerySet[Artifact, Any]:
initial = self
# These relations cause the artifacts identified by kept_id to
# be kept even if they would normally be expired, if keeper_id
# is in the set of artifacts to keep.
normal_relations = ArtifactRelation.objects.values(
keeper_id=F("artifact"), kept_id=F("target")
)
extends_relations = ArtifactRelation.objects.filter(
type=ArtifactRelation.Relations.EXTENDS
).values(keeper_id=F("target"), kept_id=F("artifact"))
relations = CTE(
normal_relations.union(extends_relations), name="relations"
)
# Follow relations recursively to establish the full set of
# artifacts to keep.
recursive = initial.values(keep_id=F("id")).union(
with_cte(
relations,
select=cte.join(
relations.queryset(), keeper_id=cte.col.keep_id
),
).values(keep_id=F("kept_id"))
)
assert isinstance(recursive, QuerySet)
return recursive
cte = CTE.recursive(keep_cte, name="keep")
keep = with_cte(cte, select=cte.join(Artifact, id=cte.col.keep_id))
assert isinstance(keep, ArtifactQuerySet)
return cast(Self, keep)
class ArtifactManager(models.Manager["Artifact"]):
"""Manager for the Artifact model."""
@override
def get_queryset(self) -> ArtifactQuerySet[Any]:
"""Use the custom QuerySet."""
return ArtifactQuerySet(self.model, using=self._db)
@classmethod
def create_from_local_artifact(
cls,
local_artifact: LocalArtifact[Any],
workspace: Workspace,
*,
created_by_work_request: Optional["WorkRequest"] = None,
) -> "Artifact":
"""Return a new Artifact based on a :py:class:`LocalArtifact`."""
artifact = Artifact.objects.create(
category=local_artifact.category,
workspace=workspace,
data=local_artifact.data.model_dump(),
created_by_work_request=created_by_work_request,
)
for artifact_path, local_path in local_artifact.files.items():
file = File.from_local_path(local_path)
file_backend = workspace.scope.upload_file_backend(file)
file_backend.add_file(local_path, fileobj=file)
FileInArtifact.objects.create(
artifact=artifact,
path=artifact_path,
file=file,
complete=True,
content_type=local_artifact.content_types.get(artifact_path),
)
return artifact
[docs]
class Artifact(models.Model):
"""Artifact model."""
category = models.CharField(
max_length=255,
# TODO: db_collation=COLLATION_CODEPOINT,
)
workspace = models.ForeignKey(Workspace, on_delete=models.PROTECT)
files = models.ManyToManyField(File, through="db.FileInArtifact")
data = models.JSONField(default=dict, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
expiration_delay = models.DurationField(blank=True, null=True)
created_by = models.ForeignKey(
"User", blank=True, null=True, on_delete=models.PROTECT
)
created_by_work_request = models.ForeignKey(
"WorkRequest", blank=True, null=True, on_delete=models.SET_NULL
)
original_artifact = models.ForeignKey(
"Artifact", blank=True, null=True, on_delete=models.SET_NULL
)
objects = ArtifactManager.from_queryset(ArtifactQuerySet)()
class Meta(TypedModelMeta):
base_manager_name = "objects"
indexes = [
# Used by debusine.server.open_metrics.DebusineCollector. This
# is imperfect (on debusine.debian.net at the time of writing it
# only takes the artifact-counting query from ~550ms to ~300ms).
# Doing any better seems likely to involve a separate
# trigger-maintained statistics table.
models.Index(
"workspace",
"category",
name="%(app_label)s_%(class)s_workspace_category",
include=["id"],
)
]
@override
def __str__(self) -> str:
"""Return basic information of Artifact."""
return (
f"Id: {self.id} "
f"Category: {self.category} "
f"Workspace: {self.workspace.id}"
)
[docs]
@override
@copy_signature_from(models.Model.save)
def save(self, **kwargs: Any) -> None:
"""Wrap save with permission checks."""
from debusine.db.context import context
if self._state.adding:
# Create
if not self.workspace.can_create_artifacts(context.pc):
user = context.user or AnonymousUser()
raise PermissionDenied(
f"{user} cannot create artifacts in {self.workspace}"
)
else:
# Update
... # TODO: check for update permissions
return super().save(**kwargs)
[docs]
def get_absolute_url(self) -> str:
"""Return the canonical URL to display the artifact."""
from debusine.server.scopes import urlconf_scope
with urlconf_scope(self.workspace.scope.name):
return reverse(
"workspaces:artifacts:detail",
kwargs={"wname": self.workspace.name, "artifact_id": self.id},
)
[docs]
def get_absolute_url_download(self) -> str:
"""Return the canonical URL to download the artifact."""
from debusine.server.scopes import urlconf_scope
with urlconf_scope(self.workspace.scope.name):
return reverse(
"workspaces:artifacts:download",
kwargs={"wname": self.workspace.name, "artifact_id": self.id},
)
[docs]
@override
def clean(self) -> None:
"""
Ensure that data is valid for this artifact category.
:raise ValidationError: for invalid data.
"""
if not isinstance(self.data, dict):
raise ValidationError({"data": "data must be a dictionary"})
try:
artifact_cls = LocalArtifact.class_from_category(self.category)
except ValueError as e:
raise ValidationError(
{"category": f"{self.category}: invalid artifact category"}
) from e
try:
artifact_cls.create_data(self.data)
except ValueError as e:
raise ValidationError(
{"data": f"invalid artifact data: {e}"}
) from e
[docs]
def work_request_can_display(self, work_request: "WorkRequest") -> bool:
"""Check if the work request can display this artifact."""
if work_request == self.created_by_work_request:
return True
task = work_request.get_task()
return self.pk in task.get_input_artifacts_ids()
[docs]
def work_request_can_change(self, work_request: "WorkRequest") -> bool:
"""Check if the work request can change this artifact."""
return work_request == self.created_by_work_request
[docs]
@permission_check(
"{user} cannot display artifact {resource.id}",
work_request=Allow.PASS,
anonymous=Allow.PASS,
)
def can_display(self, pc: PermissionContext) -> bool:
"""Check if the artifact can be displayed."""
if not self.workspace.can_display(pc):
return False
if pc.work_request is None:
return True
return self.work_request_can_display(pc.work_request)
[docs]
@permission_check(
"{user} cannot add files to artifact {resource.id}",
work_request=Allow.PASS,
)
def can_add_files(self, pc: PermissionContext) -> bool:
"""Check if files can be added to the artifact."""
if not self.workspace.can_create_artifacts(pc):
return False
if pc.work_request is None:
return True
return self.work_request_can_change(pc.work_request)
[docs]
@permission_check(
"{user} cannot add relations to artifact {resource.id}",
work_request=Allow.PASS,
)
def can_add_relation(self, pc: PermissionContext) -> bool:
"""Check if the artifact can be displayed."""
if not self.workspace.can_create_artifacts(pc):
return False
if pc.work_request is None:
return True
return self.work_request_can_change(pc.work_request)
[docs]
def create_data(self) -> ArtifactData:
"""Instantiate ArtifactData from data."""
artifact_cls = LocalArtifact.class_from_category(self.category)
artifact_data = artifact_cls.create_data(self.data)
assert isinstance(artifact_data, ArtifactData)
return artifact_data
[docs]
def get_label(self, data: ArtifactData | None = None) -> str:
"""
Return a label for this artifact.
Optionally reuse an already instantiated data model.
"""
if data is None:
data = self.create_data()
if label := data.get_label():
return label
return str(self.category)
[docs]
def effective_expiration_delay(self) -> dt.timedelta:
"""Return expiration_delay, inherited if None."""
expiration_delay = self.expiration_delay
if self.expiration_delay is None: # inherit
expiration_delay = self.workspace.default_expiration_delay
assert expiration_delay is not None
return expiration_delay
@property
def expire_at(self) -> dt.datetime | None:
"""Return computed expiration date."""
delay = self.effective_expiration_delay()
if delay == dt.timedelta(0):
return None
return self.created_at + delay
[docs]
def expired(self, at: dt.datetime) -> bool:
"""
Return True if this artifact has expired at a given datetime.
:param at: datetime to check if the artifact is expired.
:return bool: True if the artifact's expire_at is on or earlier than
the parameter at.
"""
expire_at = self.expire_at
if expire_at is None:
return False
return expire_at <= at
[docs]
def ui(self, request: "HttpRequest") -> "ArtifactUI":
"""Return a UI helper for this instance."""
from debusine.web.views.ui.artifacts import ArtifactUI
return ArtifactUI.get(request, self)
class ArtifactStatisticsManager(models.Manager["ArtifactStatistics"]):
"""Manager for ArtifactStatistics model."""
def refresh(self) -> None:
"""Refresh statistics."""
with connection.cursor() as cursor, transaction.atomic():
# The EXCLUSIVE table-level lock mode allows reads to proceed in
# parallel, but blocks writes. This is safe since this table is
# only updated by this method and only by a timer unit.
cursor.execute(
f"LOCK TABLE {self.model._meta.db_table} IN EXCLUSIVE MODE"
)
stat_fields = ("category", "scope_name")
existing_stats = {
tuple(getattr(row, field) for field in stat_fields): row
for row in self.all()
}
new_stats = {
tuple(row[field] for field in stat_fields): (
row["count"],
row["max_id"],
)
for row in Artifact.objects.annotate(
scope_name=F("workspace__scope__name")
)
.values(*stat_fields)
.annotate(count=Count("id"), max_id=Max("id"))
}
self.bulk_create(
ArtifactStatistics(
**dict(zip(stat_fields, key)),
count=new_stats[key][0],
max_id=new_stats[key][1],
)
for key in set(new_stats) - set(existing_stats)
)
to_update: list[ArtifactStatistics] = []
for key in set(existing_stats) & set(new_stats):
if (
existing_stats[key].count != new_stats[key][0]
or existing_stats[key].max_id != new_stats[key][1]
):
existing_stats[key].count = new_stats[key][0]
existing_stats[key].max_id = new_stats[key][1]
to_update.append(existing_stats[key])
self.bulk_update(to_update, ["count", "max_id"])
self.filter(
id__in={
existing_stats[key].id
for key in set(existing_stats) - set(new_stats)
}
).delete()
[docs]
class ArtifactStatistics(models.Model):
"""Artifact statistics for metrics."""
category = models.CharField(
editable=False,
# TODO: db_collation=COLLATION_CODEPOINT,
)
scope_name = models.CharField(
editable=False,
db_collation=COLLATION_PRESENTATION,
)
count = models.IntegerField()
max_id = models.BigIntegerField()
objects = ArtifactStatisticsManager()
[docs]
class FileInArtifact(models.Model):
"""File in artifact."""
artifact = models.ForeignKey(Artifact, on_delete=models.PROTECT)
path = models.CharField(
max_length=500,
db_collation=COLLATION_CODEPOINT,
)
file = models.ForeignKey(File, on_delete=models.PROTECT)
# We do a best-effort data migration here, but note that in cases where
# a file store is shared between multiple workspaces, this field may be
# left as False for files that were uploaded before this field was
# added. (debusine.debian.net only had a single workspace at the time
# of this migration, so that's not a problem there.)
complete = models.BooleanField(default=False)
# TODO: ruff is correct that we shouldn't use null=True on a CharField,
# but fixing that retroactively is non-trivial.
content_type = models.CharField( # noqa: DJ001
blank=True,
null=True,
db_collation=COLLATION_CODEPOINT,
)
class Meta(TypedModelMeta):
constraints = [
UniqueConstraint(
fields=["artifact", "path"],
name="%(app_label)s_%(class)s_unique_artifact_path",
),
]
@override
def __str__(self) -> str:
"""Return basic information of FileInArtifact."""
return (
f"Id: {self.id} Artifact: {self.artifact.id} "
f"Path: {self.path} File: {self.file.id}"
)
[docs]
def get_absolute_url(self) -> str:
"""Return an absolute URL to view this file."""
from debusine.server.scopes import urlconf_scope
with urlconf_scope(self.artifact.workspace.scope.name):
return reverse(
"workspaces:artifacts:file-detail",
kwargs={
"wname": self.artifact.workspace.name,
"artifact_id": self.artifact.id,
"path": self.path,
},
)
[docs]
def get_absolute_url_raw(self) -> str:
"""Return an absolute URL to view this file as raw."""
from debusine.server.scopes import urlconf_scope
with urlconf_scope(self.artifact.workspace.scope.name):
return reverse(
"workspaces:artifacts:file-raw",
kwargs={
"wname": self.artifact.workspace.name,
"artifact_id": self.artifact.id,
"path": self.path,
},
)
[docs]
def get_absolute_url_download(self) -> str:
"""Return an absolute URL to download this file."""
from debusine.server.scopes import urlconf_scope
with urlconf_scope(self.artifact.workspace.scope.name):
return reverse(
"workspaces:artifacts:file-download",
kwargs={
"wname": self.artifact.workspace.name,
"artifact_id": self.artifact.id,
"path": self.path,
},
)
[docs]
def ui(self, request: "HttpRequest") -> "FileInArtifactUI":
"""Return a UI helper for this instance."""
from debusine.web.views.ui.artifacts import FileInArtifactUI
return FileInArtifactUI.get(request, self)
[docs]
class FileUpload(models.Model):
"""File that is being/has been uploaded."""
file_in_artifact = models.OneToOneField(
FileInArtifact, on_delete=models.PROTECT
)
path = models.CharField(
max_length=500,
help_text="Path in the uploads directory",
unique=True,
db_collation=COLLATION_CODEPOINT,
)
last_activity_at = models.DateTimeField(auto_now_add=True)
@override
def __str__(self) -> str:
"""Return basic information."""
return f"{self.id}"
[docs]
@classmethod
def current_size(cls, artifact: Artifact, path_in_artifact: str) -> int:
"""
Return current file size.
The current file size might be smaller than the expected size of the
file if the file has not finished being uploaded.
Raise ValueError if path_in_artifact does not exist in Artifact or
if there's no FileUpload object for the specific File.
"""
try:
file_in_artifact = FileInArtifact.objects.get(
artifact=artifact, path=path_in_artifact
)
except FileInArtifact.DoesNotExist:
raise ValueError(
f'No FileInArtifact for Artifact {artifact.id} '
f'and path "{path_in_artifact}"'
)
try:
file_upload = FileUpload.objects.get(
file_in_artifact=file_in_artifact
)
except FileUpload.DoesNotExist:
raise ValueError(
f"No FileUpload for FileInArtifact {file_in_artifact.id}"
)
try:
size = file_upload.absolute_file_path().stat().st_size
except FileNotFoundError:
size = 0
return size
[docs]
@override
def delete(self, *args: Any, **kwargs: Any) -> tuple[int, dict[str, int]]:
"""Schedule deletion of the file in the store."""
file_path = self.absolute_file_path()
result = super().delete(*args, **kwargs)
# If this method is called from a transaction: transaction.on_commit()
# will call file_path.unlink when the most outer transaction is
# committed.
#
# In the case that the code is running without a transaction:
# the file_path.unlink will happen now.
#
# It's important that file_path.unlink is called only if the
# DB is updated with the deletion. Otherwise, the file could be
# deleted from the store but still referenced from the DB.
transaction.on_commit(partial(file_path.unlink, missing_ok=True))
return result
[docs]
def absolute_file_path(self) -> Path:
"""
Return the absolute file path of the file.
The files are stored in settings.DEBUSINE_UPLOAD_DIRECTORY.
"""
return Path(settings.DEBUSINE_UPLOAD_DIRECTORY) / self.path
class ArtifactRelationQuerySet[A](QuerySet["ArtifactRelation", A]):
"""Custom QuerySet for ArtifactRelation."""
def in_current_scope(self) -> "ArtifactRelationQuerySet[A]":
"""Filter to artifact relations in the current scope."""
from debusine.db.context import context
scope = context.require_scope()
return self.filter(
artifact__workspace__scope=scope, target__workspace__scope=scope
)
@permission_filter(work_request=Allow.PASS, anonymous=Allow.PASS)
def can_display(
self, pc: PermissionContext
) -> "ArtifactRelationQuerySet[A]":
"""Keep only ArtifactRelations that can be displayed."""
# Delegate to workspace can_display check
workspaces = Workspace.objects.can_display(pc)
qs = self.filter(
artifact__workspace__in=workspaces, target__workspace__in=workspaces
)
if pc.work_request is not None:
task = pc.work_request.get_task()
input_ids = task.get_input_artifacts_ids()
qs = qs.filter(
Q(artifact__pk__in=input_ids)
| Q(artifact__created_by_work_request=pc.work_request)
)
qs = qs.filter(
Q(target__pk__in=input_ids)
| Q(target__created_by_work_request=pc.work_request)
)
return qs
@permission_filter()
def can_delete(
self, pc: PermissionContext
) -> "ArtifactRelationQuerySet[A]":
"""Keep only ArtifactRelations that can be deleted."""
# Delegate to workspace OWNERS
owned_workspaces = Workspace.objects.with_role(
pc, Workspace.Roles.OWNER
)
visible_workspaces = Workspace.objects.can_display(pc)
return self.filter(
artifact__workspace__in=owned_workspaces,
target__workspace__in=visible_workspaces,
)
class ArtifactRelationManager(models.Manager["ArtifactRelation"]):
"""Manager for the ArtifactRelation model."""
@override
def get_queryset(self) -> ArtifactRelationQuerySet[Any]:
"""Use the custom QuerySet."""
return ArtifactRelationQuerySet(self.model, using=self._db)
[docs]
class ArtifactRelation(models.Model):
"""Model relations between artifacts."""
[docs]
class Relations(models.TextChoices):
EXTENDS = "extends", "Extends"
RELATES_TO = "relates-to", "Relates to"
BUILT_USING = "built-using", "Built using"
artifact = models.ForeignKey(
Artifact, on_delete=models.PROTECT, related_name="relations"
)
target = models.ForeignKey(
Artifact, on_delete=models.PROTECT, related_name="targeted_by"
)
type = models.CharField(
max_length=11,
choices=Relations.choices,
# TODO: db_collation=COLLATION_CODEPOINT,
)
objects = ArtifactRelationManager.from_queryset(ArtifactRelationQuerySet)()
class Meta(TypedModelMeta):
base_manager_name = "objects"
constraints = [
UniqueConstraint(
fields=["artifact", "target", "type"],
name="%(app_label)s_%(class)s_unique_artifact_target_type",
)
]
@override
def __str__(self) -> str:
"""Return str for the object."""
return f"{self.artifact.id} {self.type} {self.target.id}"
[docs]
@permission_check(
"{user} cannot display artifact relation {resource}",
work_request=Allow.PASS,
anonymous=Allow.PASS,
)
def can_display(self, pc: PermissionContext) -> bool:
"""Check if the artifact can be displayed."""
if not (
self.artifact.workspace.can_display(pc)
and self.target.workspace.can_display(pc)
):
return False
if pc.work_request is None:
return True
if (
pc.work_request == self.artifact.created_by_work_request
and pc.work_request == self.target.created_by_work_request
):
return True
task = pc.work_request.get_task()
input_ids = task.get_input_artifacts_ids()
return (
pc.work_request == self.artifact.created_by_work_request
or self.artifact.pk in input_ids
) and (
pc.work_request == self.target.created_by_work_request
or self.target.pk in input_ids
)
[docs]
@permission_check(
"{user} cannot delete artifact relation {resource}",
)
def can_delete(self, pc: PermissionContext) -> bool:
"""Check if the user can delete artifacts relations."""
return self.artifact.workspace.has_role(
pc, Workspace.Roles.OWNER
) and self.can_display(pc)