# 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