Source code for debusine.db.models.task_database

# 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.

"""
Database interaction for worker tasks on the server.

This TaskDatabase extension is used when Task code is run server-side with
database access.
"""

from collections.abc import Collection as AbcCollection
from typing import overload

from django.conf import settings

from debusine.artifacts.models import ArtifactCategory, CollectionCategory
from debusine.client.models import LookupChildType, RelationType
from debusine.db.models.artifacts import Artifact
from debusine.db.models.work_requests import WorkRequest
from debusine.tasks.models import LookupMultiple, LookupSingle
from debusine.tasks.server import (
    ArtifactInfo,
    MultipleArtifactInfo,
    TaskDatabaseInterface,
)


[docs] class TaskDatabase(TaskDatabaseInterface): """Implementation of database interaction in worker tasks."""
[docs] def __init__(self, work_request: WorkRequest) -> None: """Construct a :py:class:`TaskDatabase`.""" self.work_request = work_request
@staticmethod def _make_artifact_info(artifact: "Artifact") -> ArtifactInfo: """Extract basic information from an artifact.""" return ArtifactInfo( id=artifact.id, category=artifact.category, data=artifact.create_data(), ) @overload def lookup_single_artifact( self, lookup: LookupSingle, default_category: CollectionCategory | None = None, ) -> ArtifactInfo: ... @overload def lookup_single_artifact( self, lookup: None, default_category: CollectionCategory | None = None, ) -> None: ...
[docs] def lookup_single_artifact( self, lookup: LookupSingle | None, default_category: CollectionCategory | None = None, ) -> ArtifactInfo | None: """ Look up a single artifact. :param lookup: A :ref:`lookup-single`. :param default_category: If the first segment of a string lookup (which normally identifies a collection) does not specify a category, use this as the default category. :return: Information about the artifact, or None if the provided lookup is None (for convenience in some call sites). :raises KeyError: if the lookup does not resolve to an item. :raises LookupError: if the lookup is invalid in some way. """ # Import here to prevent circular imports from debusine.server.collections.lookup import lookup_single return ( None if lookup is None else self._make_artifact_info( lookup_single( lookup=lookup, workspace=self.work_request.workspace, user=self.work_request.created_by, default_category=default_category, workflow_root=self.work_request.get_workflow_root(), expect_type=LookupChildType.ARTIFACT, ).artifact ) )
[docs] def lookup_multiple_artifacts( self, lookup: LookupMultiple | None, default_category: CollectionCategory | None = None, ) -> MultipleArtifactInfo: """ Look up multiple artifacts. :param lookup: A :ref:`lookup-multiple`. :param default_category: If the first segment of a string lookup (which normally identifies a collection) does not specify a category, use this as the default category. :return: Information about each artifact. :raises KeyError: if any of the lookups does not resolve to an item. :raises LookupError: if the lookup is invalid in some way. """ # Import here to prevent circular imports from debusine.server.collections.lookup import lookup_multiple return MultipleArtifactInfo( [] if lookup is None else [ self._make_artifact_info(result.artifact) for result in lookup_multiple( lookup=lookup, workspace=self.work_request.workspace, user=self.work_request.created_by, default_category=default_category, workflow_root=self.work_request.get_workflow_root(), expect_type=LookupChildType.ARTIFACT, ) ] )
@overload def lookup_single_collection( self, lookup: LookupSingle, default_category: CollectionCategory | None = None, ) -> int: ... @overload def lookup_single_collection( self, lookup: None, default_category: CollectionCategory | None = None, ) -> None: ...
[docs] def lookup_single_collection( self, lookup: LookupSingle | None, default_category: CollectionCategory | None = None, ) -> int | None: """ Look up a single collection. :param lookup: A :ref:`lookup-single`. :param default_category: If the first segment of a string lookup (which normally identifies a collection) does not specify a category, use this as the default category. :return: The ID of the collection, or None if the provided lookup is None (for convenience in some call sites). :raises KeyError: if the lookup does not resolve to an item. :raises LookupError: if the lookup is invalid in some way. """ # Import here to prevent circular imports from debusine.server.collections.lookup import lookup_single return ( None if lookup is None else lookup_single( lookup=lookup, workspace=self.work_request.workspace, user=self.work_request.created_by, default_category=default_category, workflow_root=self.work_request.get_workflow_root(), expect_type=LookupChildType.COLLECTION, ).collection.id )
[docs] def get_server_setting(self, setting: str) -> str: """Look up a Django setting (strings only).""" value = getattr(settings, setting) assert isinstance(value, str) return value