# 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.
"""Models used by debusine client."""
import datetime as dt
import re
from collections.abc import Collection, Iterator, Sequence
from enum import StrEnum
from pathlib import Path
from typing import Annotated, Any, Literal, NewType, Self, TypeVar
import pydantic
import yaml
from debusine.artifacts.models import DebusineTaskConfiguration
from debusine.assets import AssetCategory, asset_data_model
from debusine.utils import calculate_hash
[docs]
class StrictBaseModel(pydantic.BaseModel):
"""Stricter pydantic configuration."""
model_config = pydantic.ConfigDict(validate_assignment=True)
[docs]
class Link(StrictBaseModel):
"""
Structured version of a resource link.
This is modeled as a subset of
https://developer.mozilla.org/en-US/docs/Web/HTML/Reference/Elements/link
""" # noqa: E501
#: Resource URL
href: str
type: str | None = None
[docs]
def can_be_str(self) -> bool:
"""Check if the only field set is href."""
return self.type is None
[docs]
class ServerObject(StrictBaseModel):
"""Model representing an object on the server."""
# See https://en.wikipedia.org/wiki/HATEOAS
#: URLs to related resources
links: dict[str, str | Link] = {}
[docs]
def get_link_url(self, name: str) -> str:
"""Return the href URL for a link."""
res = self.links[name]
if isinstance(res, str):
return res
return res.href
[docs]
class PaginatedResponse(StrictBaseModel):
"""Paginated response from the API."""
count: int | None = None
next: pydantic.AnyUrl | None = None
previous: pydantic.AnyUrl | None = None
results: list[dict[str, Any]]
[docs]
class WorkRequestRequest(StrictBaseModel):
"""Client send a WorkRequest to the server."""
task_name: str
workspace: str | None = None
task_data: dict[str, Any]
event_reactions: dict[str, Any]
[docs]
class WorkRequestResponse(ServerObject):
"""Server return a WorkRequest to the client."""
id: int
url: str
created_at: dt.datetime
started_at: dt.datetime | None = None
completed_at: dt.datetime | None = None
duration: float | None = None
status: str
result: str
worker: int | None = None
task_type: str
task_name: str
task_data: dict[str, Any]
dynamic_task_data: dict[str, Any] | None = None
priority_base: int
priority_adjustment: int
artifacts: list[int]
scope: str | None = None # Server >= 0.12 specifies scope
workspace: str
scheduler_tags_provided: list[str] = []
scheduler_tags_required: list[str] = []
def __str__(self) -> str:
"""Return representation of the object."""
return f'WorkRequest: {self.id}'
[docs]
class WorkRequestExternalDebsignRequest(StrictBaseModel):
"""Client sends data from an external `debsign` run to the server."""
signed_artifact: int
[docs]
class OnWorkRequestCompleted(ServerObject):
"""
Server return an OnWorkRequestCompleted to the client.
Returned via websocket consumer endpoint.
"""
work_request_id: int
completed_at: dt.datetime
result: str
[docs]
class RuntimeParameter(StrEnum):
"""WorkflowTemplate ``runtime_parameter`` special values."""
ANY = "any"
RuntimeParameters = (
dict[str, list[str] | Literal[RuntimeParameter.ANY]]
| Literal[RuntimeParameter.ANY]
)
[docs]
class WorkflowTemplateDataNew(StrictBaseModel):
"""Data for a WorkflowTemplate to be created."""
name: str
task_name: str
static_parameters: dict[str, Any]
runtime_parameters: RuntimeParameters = {}
priority: int = 0
[docs]
class WorkflowTemplateData(ServerObject):
"""Data for a WorkflowTemplate."""
id: int
name: str
task_name: str
static_parameters: dict[str, Any]
runtime_parameters: RuntimeParameters
priority: int = 0
[docs]
class CreateWorkflowRequest(StrictBaseModel):
"""Client sends a workflow creation request to the server."""
template_name: str
workspace: str | None = None
task_data: dict[str, Any]
[docs]
class FileRequest(StrictBaseModel):
"""Declare a FileRequest: client sends it to the server."""
size: pydantic.NonNegativeInt
checksums: dict[str, Annotated[str, pydantic.Field(max_length=255)]]
type: Literal["file"]
#: A media type, suitable for an HTTP ``Content-Type`` header.
content_type: str | None = None
[docs]
@staticmethod
def create_from(
path: Path, *, content_type: str | None = None
) -> "FileRequest":
"""Return a FileRequest for the file path."""
return FileRequest(
size=path.stat().st_size,
checksums={"sha256": calculate_hash(path, "sha256").hex()},
type="file",
content_type=content_type,
)
[docs]
class FileResponse(ServerObject):
"""Declare a FileResponse: server sends it to the client."""
size: pydantic.NonNegativeInt
checksums: dict[str, Annotated[str, pydantic.Field(max_length=255)]]
type: Literal["file"]
url: pydantic.AnyUrl
content_type: str | None = None
FilesRequestType = NewType("FilesRequestType", dict[str, FileRequest])
FilesResponseType = NewType("FilesResponseType", dict[str, FileResponse])
[docs]
class ArtifactCreateRequest(StrictBaseModel):
"""Declare an ArtifactCreateRequest: client sends it to the server."""
category: str
workspace: str | None = None
files: FilesRequestType = FilesRequestType({})
data: dict[str, Any] = {}
work_request: int | None = None
expire_at: dt.datetime | None = None
[docs]
class ArtifactResponse(ServerObject):
"""Declare an ArtifactResponse: server sends it to the client."""
id: int
url: str
scope: str
workspace: str
category: str
created_at: dt.datetime
data: dict[str, Any]
download_tar_gz_url: pydantic.AnyUrl
files_to_upload: list[str]
expire_at: dt.datetime | None = None
files: FilesResponseType = FilesResponseType({})
[docs]
class RemoteArtifact(ServerObject):
"""Declare RemoteArtifact."""
id: int
url: str
workspace: str
[docs]
class AssetCreateRequest(StrictBaseModel):
"""Request for the Asset creation API."""
category: AssetCategory
data: dict[str, Any]
work_request: int | None = None
workspace: str
[docs]
class AssetResponse(ServerObject):
"""Response from an Asset creation / listing API."""
id: int
category: AssetCategory
data: dict[str, Any]
work_request: int | None = None
workspace: str
[docs]
@pydantic.model_validator(mode="after")
def validate_data(self) -> Self:
"""Validate data using the correct data model."""
asset_data_model(self.category, self.data)
return self
[docs]
class AssetsResponse(
pydantic.RootModel[Sequence[AssetResponse]], StrictBaseModel
):
"""A response from the server with multiple AssetResponse objects."""
model_config = pydantic.ConfigDict(validate_assignment=True)
def __iter__(self) -> Iterator[AssetResponse]: # type: ignore[override]
"""Iterate over individual asset responses."""
return iter(self.root)
[docs]
class AssetPermissionCheckResponse(ServerObject):
"""A response from the asset permission check endpoint."""
has_permission: bool
username: str | None
user_id: int | None
resource: dict[str, Any] | None
[docs]
class RelationType(StrEnum):
"""Possible values for `RelationCreateRequest.type`."""
EXTENDS = "extends"
RELATES_TO = "relates-to"
BUILT_USING = "built-using"
[docs]
class RelationCreateRequest(StrictBaseModel):
"""Declare a RelationCreateRequest: client sends it to the server."""
artifact: int
target: int
type: RelationType
[docs]
class RelationResponse(ServerObject):
"""Declare a RelationResponse."""
id: int
artifact: int
target: int
type: RelationType
[docs]
class RelationsResponse(
pydantic.RootModel[Sequence[RelationResponse]], StrictBaseModel
):
"""A response from the server with multiple RelationResponse objects."""
def __iter__(self) -> Iterator[RelationResponse]: # type: ignore[override]
"""Iterate over individual relation responses."""
return iter(self.root)
[docs]
class LookupChildType(StrEnum):
"""Possible values for `LookupDict.child_type` and `expect_type`."""
BARE = "bare"
ARTIFACT = "artifact"
ARTIFACT_OR_PROMISE = "artifact-or-promise"
COLLECTION = "collection"
ANY = "any"
[docs]
class LookupResultType(StrEnum):
"""A collection item type returned by a lookup."""
BARE = "b"
ARTIFACT = "a"
COLLECTION = "c"
[docs]
class LookupSingleRequest(StrictBaseModel):
"""A request from the client to look up a single collection item."""
lookup: int | str
work_request: int
expect_type: LookupChildType
default_category: str | None = None
[docs]
class LookupMultipleRequest(StrictBaseModel):
"""A request from the client to look up multiple collection items."""
lookup: list[int | str | dict[str, Any]]
work_request: int
expect_type: LookupChildType
default_category: str | None = None
[docs]
class LookupSingleResponse(ServerObject):
"""A response from the server with a single lookup result."""
result_type: LookupResultType
collection_item: int | None = None
artifact: int | None = None
collection: int | None = None
[docs]
class LookupSingleResponseArtifact(LookupSingleResponse):
"""
A response from the server with a single lookup result for an artifact.
Used to assist type annotations.
"""
result_type: Literal[LookupResultType.ARTIFACT]
artifact: int
[docs]
class LookupSingleResponseCollection(LookupSingleResponse):
"""
A response from the server with a single lookup result for a collection.
Used to assist type annotations.
"""
result_type: Literal[LookupResultType.COLLECTION]
collection: int
# TODO: We should use PEP 695 type parameter declarations here instead, but
# that causes "Overloaded function implementation cannot produce return
# type" errors on Debusine.lookup_multiple.
LSR = TypeVar("LSR", bound=LookupSingleResponse, covariant=True)
[docs]
class LookupMultipleResponse(
pydantic.RootModel[Sequence[LSR]], StrictBaseModel
):
"""A response from the server with multiple lookup results."""
def __iter__(self) -> Iterator[LSR]: # type: ignore[override]
"""Iterate over individual results."""
return iter(self.root)
re_nonce = re.compile(r"^[A-Za-z0-9_-]{8,64}$")
re_challenge = re.compile(r"^\w{4,10}(?: \w{4,10}){2,7}$", re.ASCII)
[docs]
class EnrollPayload(pydantic.BaseModel):
"""Client-provided enrollment payload."""
model_config = pydantic.ConfigDict(validate_assignment=True, extra="forbid")
#: Nonce identifying the client enrollment
nonce: Annotated[str, pydantic.Field(pattern=re_nonce)]
#: Human-readable challenge to verify the client in the web UI
challenge: Annotated[str, pydantic.Field(pattern=re_challenge)]
#: Scope. Informational, the confirmation page will warn if invalid
scope: str
#: Hostname. Informational, shown in the confirmation page and in the
#: generated token comment
hostname: str
re_token = re.compile(r"^[A-Za-z0-9]{8,64}$")
[docs]
class EnrollOutcome(StrEnum):
"""User action in response to an enroll confirmation request."""
CONFIRM = "confirm"
CANCEL = "cancel"
[docs]
class EnrollConfirmPayload(pydantic.BaseModel):
"""Enrollment response from the server."""
model_config = pydantic.ConfigDict(validate_assignment=True, extra="forbid")
outcome: EnrollOutcome
token: Annotated[str, pydantic.Field(pattern=re_token)] | None = None
[docs]
class TaskConfigurationCollection(ServerObject):
"""Response fragment identifying a TaskConfiguration collection."""
id: int
name: str
data: dict[str, Any]
[docs]
class TaskConfigurationCollectionContents(ServerObject):
"""Bundle together a TaskConfiguration collection and its items."""
collection: TaskConfigurationCollection
items: list[DebusineTaskConfiguration]
[docs]
class TaskConfigurationCollectionUpdateResults(ServerObject):
"""Results of a task configuration collection update."""
added: int
updated: int
removed: int
unchanged: int
[docs]
class WorkspaceInheritanceChainElement(StrictBaseModel):
"""An element in an inheritance chain."""
id: int | None = None
scope: str | None = None
workspace: str | None = None
[docs]
@pydantic.model_validator(mode="after")
def not_empty(self) -> Self:
"""Ensure at least one of id or workspace is set."""
if self.id is None and self.workspace is None:
raise ValueError("at least id or workspace need to be set")
return self
[docs]
def matches(self, element: "WorkspaceInheritanceChainElement") -> bool:
"""Check if the components of element that are set match this one."""
return (
(element.id is None or element.id == self.id)
and (element.scope is None or element.scope == self.scope)
and (
element.workspace is None or element.workspace == self.workspace
)
)
[docs]
@classmethod
def from_string(cls, value: str) -> Self:
"""Resolve a command line argument to a WorkspaceInheritanceChain."""
if value.isdigit():
return cls(id=int(value))
elif "/" in value:
scope, workspace = value.split("/", 1)
return cls(scope=scope, workspace=workspace)
else:
return cls(workspace=value)
[docs]
class WorkspaceInheritanceChain(ServerObject):
"""Ordered list of workspaces inherited by a workspace."""
chain: list[WorkspaceInheritanceChainElement] = []
def __add__(self, other: "WorkspaceInheritanceChain") -> Self:
"""Return the concatenation of this chain and another."""
return self.__class__(chain=self.chain + other.chain)
def __sub__(self, other: "WorkspaceInheritanceChain") -> Self:
"""Return a copy of this chain without the elements in other."""
filtered: list[WorkspaceInheritanceChainElement] = []
for element in self.chain:
for to_remove in other.chain:
if element.matches(to_remove):
break
else:
filtered.append(element)
return self.__class__(chain=filtered)
[docs]
@classmethod
def from_strings(cls, values: Collection[str]) -> Self:
"""Build an inheritance chain from a list of user-provided strings."""
return cls(
chain=[
WorkspaceInheritanceChainElement.from_string(value)
for value in values
]
)
[docs]
class CollectionDataNew(StrictBaseModel):
"""Data for a Collection to be created."""
name: str
category: str
full_history_retention_period: int | None = None
metadata_only_retention_period: int | None = None
data: dict[str, Any]
[docs]
class CollectionData(ServerObject):
"""Data for a Collection."""
id: int
name: str
category: str
full_history_retention_period: int | None = None
metadata_only_retention_period: int | None = None
data: dict[str, Any]
stats: list[dict[str, Any]] | None = None
[docs]
class EmitMetricRequest(StrictBaseModel):
"""A request for the server to emit a metric on behalf of the client."""
metric_type: Literal["counter", "gauge", "summary", "histogram"]
name: str
labels: dict[str, str]
value: float
# Workarounds for https://github.com/yaml/pyyaml/issues/722
yaml.SafeDumper.add_multi_representer(
StrEnum,
yaml.representer.SafeRepresenter.represent_str,
)