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, override

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.collections import Collection
from debusine.db.models.work_requests import WorkRequest
from debusine.tasks.models import (
    LookupMultiple,
    LookupSingle,
)
from debusine.tasks.server import (
    ArtifactInfo,
    CollectionInfo,
    MultipleArtifactInfo,
    TaskDatabaseInterface,
)




class TaskDatabase(TaskDatabaseInterface):
    """Implementation of database interaction in worker tasks."""



    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(),
        )

    @staticmethod
    def _make_collection_info(collection: "Collection") -> CollectionInfo:
        """Extract basic information from a collection."""
        return CollectionInfo(
            id=collection.id,
            scope_name=collection.workspace.scope.name,
            workspace_name=collection.workspace.name,
            category=collection.category,
            name=collection.name,
            data=collection.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: ...



    @override
    def lookup_single_artifact(
        self,
        lookup: LookupSingle | None,
        default_category: CollectionCategory | None = None,
    ) -> ArtifactInfo | None:
        return (
            None
            if lookup is None
            else self._make_artifact_info(
                self.work_request.lookup_single(
                    lookup=lookup,
                    default_category=default_category,
                    expect_type=LookupChildType.ARTIFACT,
                ).artifact
            )
        )




    @override
    def lookup_multiple_artifacts(
        self,
        lookup: LookupMultiple | None,
        default_category: CollectionCategory | None = None,
    ) -> MultipleArtifactInfo:
        return MultipleArtifactInfo(
            []
            if lookup is None
            else [
                self._make_artifact_info(result.artifact)
                for result in self.work_request.lookup_multiple(
                    lookup=lookup,
                    default_category=default_category,
                    expect_type=LookupChildType.ARTIFACT,
                )
            ]
        )




    @override
    def find_related_artifacts(
        self,
        artifact_ids: AbcCollection[int],
        target_category: ArtifactCategory,
        relation_type: RelationType = RelationType.RELATES_TO,
    ) -> MultipleArtifactInfo:
        return MultipleArtifactInfo(
            [
                self._make_artifact_info(artifact)
                for artifact in Artifact.objects.filter(
                    targeted_by__artifact__in=artifact_ids,
                    category=target_category,
                    targeted_by__type=relation_type,
                )
            ]
        )


    @overload
    def lookup_single_collection(
        self,
        lookup: LookupSingle,
        default_category: CollectionCategory | None = None,
    ) -> CollectionInfo: ...

    @overload
    def lookup_single_collection(
        self,
        lookup: None,
        default_category: CollectionCategory | None = None,
    ) -> None: ...



    @override
    def lookup_single_collection(
        self,
        lookup: LookupSingle | None,
        default_category: CollectionCategory | None = None,
    ) -> CollectionInfo | None:
        return (
            None
            if lookup is None
            else self._make_collection_info(
                self.work_request.lookup_single(
                    lookup=lookup,
                    default_category=default_category,
                    expect_type=LookupChildType.COLLECTION,
                ).collection
            )
        )




    @override
    def get_server_setting(self, setting: str) -> str:
        value = getattr(settings, setting)
        assert isinstance(value, str)
        return value




    @override
    def export_suite_signing_keys(self, suite_id: int) -> str | None:
        # Import here to prevent circular imports
        from debusine.server.collections import DebianSuiteManager

        suite = Collection.objects.get(id=suite_id)
        assert isinstance(suite.manager, DebianSuiteManager)
        return suite.manager.export_signing_keys()