Source code for debusine.db.models.workspaces

# 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}"