# 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.db import models
from django.db.models import UniqueConstraint
from debusine.db.models.files import File, FileStore
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_with_name(cls, name: str, **kwargs: Any) -> "Workspace":
"""Return a new Workspace with name and the default FileStore."""
kwargs.setdefault("default_file_store", FileStore.default())
return Workspace.objects.create(name=name, **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, unique=True)
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",
)
[docs] def is_file_in_workspace(self, fileobj: File) -> bool:
"""Return True if fileobj is in any store available for Workspace."""
file_stores = [self.default_file_store, *self.other_file_stores.all()]
for file_store in file_stores:
if file_store.fileinstore_set.filter(file=fileobj).exists():
return True
return False
[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}"