Python client API

The Debusine class

class debusine.client.debusine.Debusine(base_api_url: str, api_token: str | None = None, *, logger: Logger)

Bases: object

Class to interact with debusine server.

API_VERSION = '1.0'

__init__(base_api_url: str, api_token: str | None = None, *, logger: Logger)

Initialize client.

Parameters:

• base_api_url – URL for API endpoint (e.g. http://localhost/api)

• api_token – optional token to be used for the calls.

artifact_create(artifact: LocalArtifact[Any], *, workspace: str | None, work_request: int | None = None, expire_at: datetime.datetime | None = None) → ArtifactResponse

Create artifact in the debusine server.

artifact_get(artifact_id: int) → ArtifactResponse

Get artifact information.

Use download_artifact() to download the artifact.

download_artifact(artifact_id: int, destination: Path, *, tarball: bool = False) → ArtifactResponse

Download artifact_id into destination directory.

Parameters:

• artifact_id – artifact id to download

• destination – destination directory to download/uncompress

• tarball – True to only download the tarball (artifact-id.tar.gz), False to uncompress it

download_artifact_file(artifact_id: int, path_in_artifact: str, destination: Path) → ArtifactResponse

Download artifact_id into destination directory.

Parameters:

• artifact_id – artifact id to download

• path_in_artifact – download the file from the artifact with this name

• destination – destination file to download to

lookup_multiple(lookup: CollectionItemLookupMultiple, work_request: int, expect_type: ~typing.Literal[<CollectionItemType.ARTIFACT: 'a'>], default_category: str | None = None) → LookupMultipleResponse[LookupSingleResponseArtifact]

lookup_multiple(lookup: CollectionItemLookupMultiple, work_request: int, expect_type: ~typing.Literal[<CollectionItemType.COLLECTION: 'c'>], default_category: str | None = None) → LookupMultipleResponse[LookupSingleResponseCollection]

lookup_multiple(lookup: CollectionItemLookupMultiple, work_request: int, expect_type: CollectionItemType, default_category: str | None = None) → LookupMultipleResponse[LookupSingleResponse]

Look up multiple collection items.

lookup_single(lookup: CollectionItemLookupSingle, work_request: int, expect_type: ~typing.Literal[<CollectionItemType.ARTIFACT: 'a'>], default_category: str | None = None) → LookupSingleResponseArtifact

lookup_single(lookup: CollectionItemLookupSingle, work_request: int, expect_type: ~typing.Literal[<CollectionItemType.COLLECTION: 'c'>], default_category: str | None = None) → LookupSingleResponseCollection

lookup_single(lookup: CollectionItemLookupSingle, work_request: int, expect_type: CollectionItemType, default_category: str | None = None) → LookupSingleResponse

Look up a single collection item.

on_work_request_completed(*, workspaces: list[str] | None = None, last_completed_at: pathlib.Path | None = None, command: str, working_directory: Path)

Execute command when a work request is completed.

async static process_async_message(msg: WSMessage, msg_text_to_callable: dict[str, debusine.client.debusine._MessageProcessor], logger: Logger) → bool

Process “msg”: logs possible error, raise errors.

If msg.type is aiohttp.WSMsgType.TEXT: decode msg.data (contains JSON), call callable msg_text_to_callable[msg.data[“text”]](msg_content: dict).

Return True a callable from msg_to_callable is called or False if not (invalid messages, etc.).

Raises:

TokenDisabledError – if reason_code is TOKEN_DISABLED.

relation_create(artifact_id: int, target_id: int, relation_type: Literal['extends', 'relates-to', 'built-using']) → RelationResponse

Create a new relation between artifacts.

Parameters:

• artifact_id – relation from

• target_id – relation to

• relation_type – type of relation such as extends, relates-to, built-using

Returns:

True if the relation already existed/has been created, False if it could not be created

upload_artifact(local_artifact: LocalArtifact[Any], *, workspace: str | None, work_request: int | None = None, expire_at: datetime.datetime | None = None) → RemoteArtifact

Upload (create and upload files) the local_artifact to the server.

upload_files(artifact_id: int, upload_files: dict[str, pathlib.Path], base_directory: pathlib.Path | None = None) → None

Upload into artifact the files.

Parameters:

• artifact_id – artifact_id to upload files to.

• upload_files – list of files to upload.

• base_directory – base directory for relative path’s files to upload.

static work_request_completed_path(work_request_id: int) → str

Return path to update the completed result for work_request_id.

work_request_completed_update(work_request_id: int, result: str)

Update work_request_id as completed with result.

work_request_create(work_request: WorkRequestRequest) → WorkRequestResponse

Create a work request (via POST /work-request/).

Returns:

WorkRequest returned by the server.

Raises:

see _api_request method documentation.

work_request_get(work_request_id: int) → WorkRequestResponse

Get WorkRequest for work_request_id.

Parameters:

work_request_id – id to fetch the status of.

Raises:

many – see _api_request method documentation.

work_request_iter() → Iterable[WorkRequestResponse]

List all WorkRequests.

Raises:

many – see _api_request method documentation.

work_request_update(work_request_id: int, *, priority_adjustment: int) → None

Update properties of work_request_id.

static write_last_completed_at(completed_at_file: pathlib.Path | None, completed_at: datetime.datetime | None) → None

Write to completed_at_file the completed_at datetime.

Associated data models

The parameters of the HTTP requests and the responses of the server are represented by Pydantic models that are described below.

Models used by debusine client.

class debusine.client.models.ArtifactCreateRequest(*, category: str, workspace: str | None = None, files: FilesRequestType = {}, data: dict[str, Any] = {}, work_request: int | None = None, expire_at: datetime.datetime | None = None)

Bases: StrictBaseModel

Declare an ArtifactCreateRequest: client sends it to the server.

category: str

data: dict[str, Any]

expire_at: datetime.datetime | None

files: FilesRequestType

work_request: int | None

workspace: str | None

class debusine.client.models.ArtifactResponse(*, id: int, workspace: str, category: str, created_at: datetime, data: dict[str, Any], download_tar_gz_url: AnyUrl, files_to_upload: list[str], expire_at: datetime.datetime | None = None, files: FilesResponseType = {})

Bases: StrictBaseModel

Declare an ArtifactResponse: server sends it to the client.

category: str

created_at: datetime

data: dict[str, Any]

download_tar_gz_url: AnyUrl

expire_at: datetime.datetime | None

files: FilesResponseType

files_to_upload: list[str]

id: int

workspace: str

class debusine.client.models.CollectionItemType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: StrEnum

A collection item type.

ARTIFACT = 'a'

BARE = 'b'

COLLECTION = 'c'

class debusine.client.models.FileRequest(*, size: ConstrainedIntValue, checksums: dict[str, typing.Annotated[str, FieldInfo(default=PydanticUndefined, max_length=255, extra={})]], type: Literal['file'])

Bases: StrictBaseModel

Declare a FileRequest: client sends it to the server.

checksums: dict[str, typing.Annotated[str, FieldInfo(default=PydanticUndefined, max_length=255, extra={})]]

static create_from(path: Path) → FileRequest

Return a FileRequest for the file path.

size: int

type: Literal['file']

class debusine.client.models.FileResponse(*, size: ConstrainedIntValue, checksums: dict[str, typing.Annotated[str, FieldInfo(default=PydanticUndefined, max_length=255, extra={})]], type: Literal['file'], url: AnyUrl)

Bases: StrictBaseModel

Declare a FileResponse: server sends it to the client.

checksums: dict[str, typing.Annotated[str, FieldInfo(default=PydanticUndefined, max_length=255, extra={})]]

size: int

type: Literal['file']

url: AnyUrl

class debusine.client.models.LookupMultipleRequest(*, lookup: list[int | str | dict[str, Any]], work_request: int, expect_type: CollectionItemType, default_category: str | None = None)

Bases: StrictBaseModel

A request from the client to look up multiple collection items.

default_category: str | None

expect_type: CollectionItemType

lookup: list[int | str | dict[str, Any]]

work_request: int

class debusine.client.models.LookupMultipleResponse(*, __root__: Sequence[LSR])

Bases: StrictBaseModel, Generic[LSR]

A response from the server with multiple lookup results.

class debusine.client.models.LookupSingleRequest(*, lookup: int | str, work_request: int, expect_type: CollectionItemType, default_category: str | None = None)

Bases: StrictBaseModel

A request from the client to look up a single collection item.

default_category: str | None

expect_type: CollectionItemType

lookup: int | str

work_request: int

class debusine.client.models.LookupSingleResponse(*, result_type: CollectionItemType, collection_item: int | None = None, artifact: int | None = None, collection: int | None = None)

Bases: StrictBaseModel

A response from the server with a single lookup result.

artifact: int | None

collection: int | None

collection_item: int | None

result_type: CollectionItemType

class debusine.client.models.LookupSingleResponseArtifact(*, result_type: ~typing.Literal[<CollectionItemType.ARTIFACT: 'a'>], collection_item: int | None = None, artifact: int, collection: int | None = None)

Bases: LookupSingleResponse

A response from the server with a single lookup result for an artifact.

Used to assist type annotations.

artifact: int

result_type: ARTIFACT: 'a'>]

class debusine.client.models.LookupSingleResponseCollection(*, result_type: ~typing.Literal[<CollectionItemType.COLLECTION: 'c'>], collection_item: int | None = None, artifact: int | None = None, collection: int)

Bases: LookupSingleResponse

A response from the server with a single lookup result for a collection.

Used to assist type annotations.

collection: int

result_type: COLLECTION: 'c'>]

class debusine.client.models.OnWorkRequestCompleted(*, work_request_id: int, completed_at: datetime, result: str)

Bases: StrictBaseModel

Server return an OnWorkRequestCompleted to the client.

Returned via websocket consumer endpoint.

completed_at: datetime

result: str

work_request_id: int

class debusine.client.models.PaginatedResponse(*, count: int | None = None, next: pydantic.networks.AnyUrl | None = None, previous: pydantic.networks.AnyUrl | None = None, results: list[dict[str, Any]])

Bases: StrictBaseModel

Paginated response from the API.

count: int | None

next: pydantic.networks.AnyUrl | None

previous: pydantic.networks.AnyUrl | None

results: list[dict[str, Any]]

class debusine.client.models.RelationCreateRequest(*, artifact: int, target: int, type: Literal['extends', 'relates-to', 'built-using'])

Bases: StrictBaseModel

Declare a RelationCreateRequest: client sends it to the server.

artifact: int

target: int

type: Literal['extends', 'relates-to', 'built-using']

class debusine.client.models.RelationResponse(*, artifact: int, target: int, type: Literal['extends', 'relates-to', 'built-using'], id: int)

Bases: RelationCreateRequest

Declare a RelationResponse.

id: int

class debusine.client.models.RemoteArtifact(*, id: int, workspace: str)

Bases: StrictBaseModel

Declare RemoteArtifact.

id: int

workspace: str

class debusine.client.models.StrictBaseModel

Bases: BaseModel

Stricter pydantic configuration.

class Config

Bases: object

Set up stricter pydantic Config.

validate_assignment = True

class debusine.client.models.WorkRequestRequest(*, task_name: str, workspace: str | None = None, task_data: dict[str, Any], event_reactions: dict[str, Any])

Bases: StrictBaseModel

Client send a WorkRequest to the server.

event_reactions: dict[str, Any]

task_data: dict[str, Any]

task_name: str

workspace: str | None

class debusine.client.models.WorkRequestResponse(*, id: int, created_at: datetime, started_at: datetime.datetime | None = None, completed_at: datetime.datetime | None = None, duration: int | None = None, status: str, result: str, worker: int | None = None, task_name: str, task_data: dict[str, Any], priority_base: int, priority_adjustment: int, artifacts: list[int], workspace: str)

Bases: StrictBaseModel

Server return a WorkRequest to the client.

artifacts: list[int]

completed_at: datetime.datetime | None

created_at: datetime

duration: int | None

id: int

priority_adjustment: int

priority_base: int

result: str

started_at: datetime.datetime | None

status: str

task_data: dict[str, Any]

task_name: str

worker: int | None

workspace: str

debusine.client.models.model_to_json_serializable_dict(model: BaseModel) → dict[Any, Any]

Similar to model.dict() but the returned dictionary is JSON serializable.

For example, a datetime() is not JSON serializable. Using this method will return a dictionary with a string instead of a datetime object.

Representation of artifacts

Artifacts to be uploaded are first created as a debusine.artifacts.local_artifact.LocalArtifact object. There are many descendants of that class available in debusine.artifacts, one for each category of artifact.

class debusine.artifacts.local_artifact.LocalArtifact(*, category: str, files: dict[str, pathlib.Path] = None, data: AD)

Bases: BaseModel, Generic[AD], ABC

Represent an artifact locally.

class Config

Bases: object

Set up stricter pydantic Config.

extra = 'forbid'

validate_assignment = True

__init__(**kwargs)

Set category to _category by default.

add_local_file(file: Path, *, artifact_base_dir: pathlib.Path | None = None, override_name: str | None = None)

Add a local file in the artifact.

Parameters:

• file – file in the local file system that is added to the artifact

• artifact_base_dir – base directory of the artifact. Must be an absolute path. If it’s None: file is added in the root of the artifact. If it’s not None: file is added with the relative path of the file with the artifact_base_dir. E.g. file=/tmp/artifact/dir1/file1 artifact_base_dir=/tmp/artifact Path of this file in the artifact: dir1/file1

• override_name – if not None: use it instead of file.name

Raises:

ValueError – artifact_base_dir is not absolute or is not a directory; file does not exist; the path in the artifact already had a file.

static artifact_categories() → list[str]

Return list of artifact categories.

category: str

Artifact type

static class_from_category(category: str) → type[debusine.artifacts.local_artifact.LocalArtifact[Any]]

Return class sub_local_artifact.

classmethod create_data(data_dict: dict[str, Any]) → AD

Instantiate a data model from a dict.

data: AD

Artifact data

files: dict[str, pathlib.Path]

serialize_for_create_artifact(*, workspace: str | None, work_request: int | None = None, expire_at: datetime.datetime | None = None) → dict[str, Any]

Return dictionary to be used by the API to create an artifact.

validate_model()

Raise ValueError with an error if the model is not valid.

Debusine client.

class debusine.artifacts.AutopkgtestArtifact(*, category: str, files: dict[str, pathlib.Path] = None, data: AD)

Bases: LocalArtifact[DebianAutopkgtest]

Autopkgtest: encapsulate result of the Autopkgtest run.

classmethod create(artifact_directory: Path, data: DebianAutopkgtest) → Self

Return AutopkgtestArtifact with the files and data set.

class debusine.artifacts.BinaryPackage(*, category: str, files: dict[str, pathlib.Path] = None, data: AD)

Bases: LocalArtifact[DebianBinaryPackage]

BinaryPackage: encapsulates a single *.deb / *.udeb.

classmethod create(*, file: Path) → Self

Return a BinaryPackage setting file and data.

classmethod files_exactly_one(files: dict[str, pathlib.Path]) → dict[str, pathlib.Path]

Raise ValueError if len(files) != 1.

classmethod files_must_end_in_deb_or_udeb(files: dict[str, pathlib.Path]) → dict[str, pathlib.Path]

Raise ValueError if a file does not end in .deb or .udeb.

class debusine.artifacts.BinaryPackages(*, category: str, files: dict[str, pathlib.Path] = None, data: AD)

Bases: LocalArtifact[DebianBinaryPackages]

BinaryPackages: encapsulates a group of *.deb / *.udeb.

classmethod create(*, srcpkg_name: str, srcpkg_version: str, version: str, architecture: str, files: list[pathlib.Path]) → Self

Return a BinaryPackages setting files and data.

classmethod files_more_than_zero(files: dict[str, pathlib.Path]) → dict[str, pathlib.Path]

Raise ValueError if len(files) == 0.

classmethod files_must_end_in_deb_or_udeb(files: dict[str, pathlib.Path]) → dict[str, pathlib.Path]

Raise ValueError if a file does not end in .deb or .udeb.

class debusine.artifacts.BlhcArtifact(*, category: str, files: dict[str, pathlib.Path] = None, data: EmptyArtifactData = None)

Bases: LocalArtifact[EmptyArtifactData]

BlhcArtifact: encapsulate result of the blhc run.

classmethod create(blhc_output: Path) → Self

Return a BlhcArtifact with the files set.

data: EmptyArtifactData

Artifact data

class debusine.artifacts.DebianSystemTarballArtifact(*, category: str, files: dict[str, pathlib.Path] = None, data: AD)

Bases: LocalArtifact[DebianSystemTarball]

Contain system.tar.xz file with a Debian.

Can be used by a chroot, container, etc.

classmethod create(tarball: Path, data: dict[str, Any]) → Self

Return a DebianSystemTarballArtifact with the tarball file.

class debusine.artifacts.LintianArtifact(*, category: str, files: dict[str, pathlib.Path] = None, data: AD)

Bases: LocalArtifact[DebianLintian]

LintianArtifact: encapsulate result of the Lintian run.

classmethod create(analysis: Path, lintian_output: Path, summary: DebianLintianSummary) → Self

Return a LintianArtifact with the files set.

class debusine.artifacts.LocalArtifact(*, category: str, files: dict[str, pathlib.Path] = None, data: AD)

Bases: BaseModel, Generic[AD], ABC

Represent an artifact locally.

class Config

Bases: object

Set up stricter pydantic Config.

extra = 'forbid'

validate_assignment = True

__init__(**kwargs)

Set category to _category by default.

add_local_file(file: Path, *, artifact_base_dir: pathlib.Path | None = None, override_name: str | None = None)

Add a local file in the artifact.

Parameters:

• file – file in the local file system that is added to the artifact

• artifact_base_dir – base directory of the artifact. Must be an absolute path. If it’s None: file is added in the root of the artifact. If it’s not None: file is added with the relative path of the file with the artifact_base_dir. E.g. file=/tmp/artifact/dir1/file1 artifact_base_dir=/tmp/artifact Path of this file in the artifact: dir1/file1

• override_name – if not None: use it instead of file.name

Raises:

ValueError – artifact_base_dir is not absolute or is not a directory; file does not exist; the path in the artifact already had a file.

static artifact_categories() → list[str]

Return list of artifact categories.

category: str

Artifact type

static class_from_category(category: str) → type[debusine.artifacts.local_artifact.LocalArtifact[Any]]

Return class sub_local_artifact.

classmethod create_data(data_dict: dict[str, Any]) → AD

Instantiate a data model from a dict.

data: AD

Artifact data

files: dict[str, pathlib.Path]

serialize_for_create_artifact(*, workspace: str | None, work_request: int | None = None, expire_at: datetime.datetime | None = None) → dict[str, Any]

Return dictionary to be used by the API to create an artifact.

validate_model()

Raise ValueError with an error if the model is not valid.

class debusine.artifacts.PackageBuildLog(*, category: str, files: dict[str, pathlib.Path] = None, data: AD)

Bases: LocalArtifact[DebianPackageBuildLog]

PackageBuildLog: represents a build log file.

classmethod create(*, file: Path, source: str, version: str) → Self

Return a PackageBuildLog.

classmethod file_must_end_in_build(files: dict[str, pathlib.Path]) → dict[str, pathlib.Path]

Raise ValueError if the file does not end in .build.

classmethod validate_files_length_is_one(files: dict[str, pathlib.Path]) → dict[str, pathlib.Path]

Validate that artifact has only one file.

class debusine.artifacts.SourcePackage(*, category: str, files: dict[str, pathlib.Path] = None, data: AD)

Bases: LocalArtifact[DebianSourcePackage]

SourcePackage: contains source code to be built into BinaryPackages.

classmethod create(*, name: str, version: str, files: list[pathlib.Path]) → Self

Return a SourcePackage setting files and data.

classmethod files_contain_one_dsc(files: dict[str, pathlib.Path])

Raise ValueError when files does not have exactly 1 .dsc file.

classmethod files_contains_files_in_dsc(files: dict[str, pathlib.Path])

Validate that set(files) == set(files_in_dsc_file).

Exception: The .dsc file must be in files but not in the .dsc file.

class debusine.artifacts.Upload(*, category: str, files: dict[str, pathlib.Path] = None, data: AD)

Bases: LocalArtifact[DebianUpload]

Upload: encapsulate a .changes and files listed in it.

classmethod create(*, changes_file: Path, exclude_files: frozenset[pathlib.Path] | set[pathlib.Path] = frozenset({})) → Self

Return a Upload. Add the changes_files and files listed in it.

Parameters:

• changes_file – a .changes file. Parsed by deb822.Changes.

• exclude_files – do not add them in files even if listed in the Files section in the changes_file.

classmethod files_contain_changes(files: dict[str, pathlib.Path])

Raise ValueError when files does not have exactly 1 .changes file.

classmethod files_contains_files_in_changes(files: dict[str, pathlib.Path])

Validate that set(files) == set(files_in_changes_file).

Exception: The .changes file must be in files but not in the .changes file.

class debusine.artifacts.WorkRequestDebugLogs(*, category: str, files: dict[str, pathlib.Path] = None, data: EmptyArtifactData = None)

Bases: LocalArtifact[EmptyArtifactData]

WorkRequestDebugLogs: help debugging issues executing the task.

Log files for debusine users in order to debug possible problems in their WorkRequests.

classmethod create(*, files: Iterable[Path]) → Self

Return a WorkRequestDebugLogs.

data: EmptyArtifactData

Artifact data