# Copyright 2019, 2021-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 workspaces."""
from collections.abc import Sequence
from datetime import timedelta
from typing import Any, Optional, TYPE_CHECKING
from django.conf import settings
from django.db import models
from django.db.models import UniqueConstraint
from debusine.db.models.files import File, FileStore
from debusine.db.models.scopes import Scope
if TYPE_CHECKING:
from django_stubs_ext.db.models import TypedModelMeta
from debusine.db.models import Collection, User
else:
TypedModelMeta = object
class WorkspaceManager(models.Manager["Workspace"]):
"""Manager for Workspace model."""
@classmethod
def create_on_default_scope(cls, name: str, **kwargs: Any) -> "Workspace":
"""
Create a Workspace in the default scope with default FileStore.
This method is used during the transition to scoped workspaces, and is
intended to be deprecated and removed once scopes are in use.
"""
scope = Scope.objects.get(name=settings.DEBUSINE_DEFAULT_SCOPE)
kwargs.setdefault("default_file_store", FileStore.default())
return Workspace.objects.create(name=name, scope=scope, **kwargs)
DEFAULT_WORKSPACE_NAME = "System"
[docs]def default_workspace() -> "Workspace":
"""Return the default Workspace."""
return Workspace.objects.get(name=DEFAULT_WORKSPACE_NAME)
[docs]class Workspace(models.Model):
"""Workspace model."""
objects = WorkspaceManager()
name = models.CharField(max_length=255)
default_file_store = models.ForeignKey(
FileStore, on_delete=models.PROTECT, related_name="default_workspaces"
)
other_file_stores = models.ManyToManyField(
FileStore, related_name="other_workspaces"
)
public = models.BooleanField(default=False)
default_expiration_delay = models.DurationField(
default=timedelta(0),
help_text="minimal time that a new artifact is kept in the"
" workspace before being expired",
)
inherits = models.ManyToManyField(
"db.Workspace",
through="db.WorkspaceChain",
through_fields=("child", "parent"),
related_name="inherited_by",
)
scope = models.ForeignKey(
Scope, on_delete=models.PROTECT, related_name="workspaces"
)
class Meta(TypedModelMeta):
constraints = [
UniqueConstraint(
fields=["scope", "name"],
name="%(app_label)s_%(class)s_unique_scope_name",
),
]
[docs] def is_file_in_workspace(self, fileobj: File) -> bool:
"""Return True if fileobj is in any store available for Workspace."""
from debusine.db.models import FileInArtifact
file_stores = [self.default_file_store, *self.other_file_stores.all()]
if not any(
file_store.fileinstore_set.filter(file=fileobj).exists()
for file_store in file_stores
):
return False
if not FileInArtifact.objects.filter(
artifact__workspace=self, file=fileobj, complete=True
).exists():
return False
return True
[docs] def set_inheritance(self, chain: Sequence["Workspace"]) -> None:
"""Set the inheritance chain for this workspace."""
# Check for duplicates in the chain before altering the database
seen: set[int] = set()
for workspace in chain:
if workspace.pk in seen:
raise ValueError(
f"duplicate workspace {workspace.name!r}"
" in inheritance chain"
)
seen.add(workspace.pk)
WorkspaceChain.objects.filter(child=self).delete()
for idx, workspace in enumerate(chain):
WorkspaceChain.objects.create(
child=self, parent=workspace, order=idx
)
[docs] def get_collection(
self,
*,
user: Optional["User"],
category: str,
name: str,
visited: set[int] | None = None,
) -> "Collection":
"""
Lookup a collection by category and name.
If the collection is not found in this workspace, it follows the
workspace inheritance chain using a depth-first search.
:param user: user to use for permission checking
:param category: collection category
:param name: collection name
:param visited: for internal use only: state used during graph
traversal
:raises Collection.DoesNotExist: if the collection was not found
"""
from debusine.db.models import Collection
# Ensure that the user can access this workspace
if not self.public and user is None:
raise Collection.DoesNotExist
# Lookup in this workspace
try:
return Collection.objects.get(
workspace=self, category=category, name=name
)
except Collection.DoesNotExist:
pass
if visited is None:
visited = set()
visited.add(self.pk)
# Follow the inheritance chain
for node in self.chain_parents.order_by("order").select_related(
"parent"
):
workspace = node.parent
# Break inheritance loops
if workspace.pk in visited:
continue
try:
return workspace.get_collection(
user=user, category=category, name=name, visited=visited
)
except Collection.DoesNotExist:
pass
raise Collection.DoesNotExist
def __str__(self) -> str:
"""Return basic information of Workspace."""
return f"Id: {self.id} Name: {self.name}"
class WorkspaceChain(models.Model):
"""Workspace chaining model."""
child = models.ForeignKey(
Workspace,
on_delete=models.CASCADE,
related_name="chain_parents",
help_text="Workspace that falls back on `parent` for lookups",
)
parent = models.ForeignKey(
Workspace,
on_delete=models.CASCADE,
related_name="chain_children",
help_text="Workspace to be looked up if lookup in `child` fails",
)
order = models.IntegerField(
help_text="Lookup order of this element in the chain",
)
class Meta(TypedModelMeta):
constraints = [
UniqueConstraint(
fields=["child", "parent"],
name="%(app_label)s_%(class)s_unique_child_parent",
),
UniqueConstraint(
fields=["child", "order"],
name="%(app_label)s_%(class)s_unique_child_order",
),
]
def __str__(self) -> str:
"""Return basic information of Workspace."""
return f"{self.order}:{self.child.name}→{self.parent.name}"