# 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