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 typing import TYPE_CHECKING, overload

from django.conf import settings

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

if TYPE_CHECKING:
    from debusine.db.models.artifacts import Artifact


[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