opsicommon.messagebus package

Submodules

opsicommon.messagebus.file_transfer module

This file is part of opsi - https://www.opsi.org

class opsicommon.messagebus.file_transfer.FileDownload(file_download_request, send_message, sender='@', back_channel=None)

Bases: FileTransfer

Parameters:
async send_data()
Return type:

None

async start()
Return type:

None

class opsicommon.messagebus.file_transfer.FileTransfer(send_message, file_request, sender='@', back_channel=None)

Bases: object

Parameters:
chunk_timeout = 300
default_destination: Path | None = None
async process_file_chunk(message)
Parameters:

message (FileChunkMessage)

Return type:

None

async stop()
Return type:

None

class opsicommon.messagebus.file_transfer.FileUpload(file_upload_request, send_message, sender='@', back_channel=None)

Bases: FileTransfer

Parameters:
async start()
Return type:

None

opsicommon.messagebus.file_transfer.calc_no_of_chunks(file_size, chunk_size)
Parameters:

  • file_size (int)

  • chunk_size (int)

Return type:

int

async opsicommon.messagebus.file_transfer.process_messagebus_message(message, send_message, *, sender='@', back_channel=None)
Parameters:
Return type:

None

opsicommon.messagebus.file_transfer.remove_file_transfer(file_id)
Parameters:

file_id (str)

Return type:

None

async opsicommon.messagebus.file_transfer.stop_running_file_transfers()
Return type:

None

opsicommon.messagebus.message module

This file is part of opsi - https://www.opsi.org

class opsicommon.messagebus.message.ChannelSubscriptionEventMessage(**data)

Bases: Message

Message for confirming channel access

Is response to ChannelSubscriptionRequestMessage and contains total subscribed_channels or error.

Parameters:

data (Any)

error: Error | None
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'error': FieldInfo(annotation=Union[Error, NoneType], required=False, default=None), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'subscribed_channels': FieldInfo(annotation=list[str], required=False, default_factory=list), 'type': FieldInfo(annotation=str, required=False, default='channel_subscription_event')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

subscribed_channels: list[str]
type: str
class opsicommon.messagebus.message.ChannelSubscriptionOperation(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: StrEnum

ADD = 'add'
REMOVE = 'remove'
SET = 'set'
class opsicommon.messagebus.message.ChannelSubscriptionRequestMessage(**data)

Bases: Message

Message for requesting channel access

Can be used to set, add or remove subscribed channels.

Parameters:

data (Any)

channels: list[str]
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'channels': FieldInfo(annotation=list[str], required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'operation': FieldInfo(annotation=str, required=False, default='set'), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=False, default='channel_subscription_request')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

operation: str
type: str
class opsicommon.messagebus.message.Error(**data)

Bases: BaseModel

Parameters:

data (Any)

code: ErrorCode | Annotated[int, AfterValidator(lambda x: None)] | None
details: Any
message: str
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'code': FieldInfo(annotation=Union[ErrorCode, Annotated[int, AfterValidator], NoneType], required=False, default=None), 'details': FieldInfo(annotation=Any, required=False, default=None), 'message': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class opsicommon.messagebus.message.ErrorCode(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: StrEnum

FILE_NOT_FOUND = 'file_not_found'
PERMISSION_ERROR = 'permission_error'
TIMEOUT_REACHED = 'timeout_reached'
class opsicommon.messagebus.message.EventMessage(**data)

Bases: Message

Class for Event Messages

Used to notify messagebus of an event that occured.

Parameters:

data (Any)

data: dict[str, Any]
event: str
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'data': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'event': FieldInfo(annotation=str, required=True), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=False, default='event')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str
class opsicommon.messagebus.message.FileChunkMessage(**data)

Bases: FileTransferMessage

Message to send a chunk of a file

Contains the chunk number (for ordering in assembly) and the actual data as bytes. The last chunk of a file should contain last=True to conclude the upload.

Parameters:

data (Any)

data: bytes
last: bool
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'data': FieldInfo(annotation=bytes, required=True), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'file_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'last': FieldInfo(annotation=bool, required=False, default=False), 'number': FieldInfo(annotation=int, required=True), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=False, default='file_chunk')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

number: int
type: str
class opsicommon.messagebus.message.FileDownloadRequestMessage(**data)

Bases: FileTransferMessage

Message for requesting a file download

Parameters:

data (Any)

chunk_size: int | None
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'chunk_size': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'file_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'path': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'terminal_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'type': FieldInfo(annotation=str, required=False, default=<MessageType.FILE_DOWNLOAD_REQUEST: 'file_download_request'>)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

path: str | None
terminal_id: str | None
type: str
class opsicommon.messagebus.message.FileDownloadResponseMessage(**data)

Bases: FileTransferMessage

Message with information like file size, type, number of chunks.

Parameters:

data (Any)

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'file_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'no_of_chunks': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'size': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'type': FieldInfo(annotation=str, required=False, default=<MessageType.FILE_DOWNLOAD_INFORMATION: 'file_download_information'>)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

no_of_chunks: int | None
size: int | None
type: str
class opsicommon.messagebus.message.FileTransferErrorMessage(**data)

Bases: FileTransferMessage

Message reporting a file transfer related Error

Used to transport Error object via messagebus.

Parameters:

data (Any)

error: Error
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'error': FieldInfo(annotation=Error, required=True), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'file_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=False, default='file_transfer_error')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str
class opsicommon.messagebus.message.FileTransferMessage(**data)

Bases: Message, ABC

Message regarding a file transfer.

Contains a unique file_id.

Parameters:

data (Any)

file_id: UUID4Str
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'file_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class opsicommon.messagebus.message.FileUploadRequestMessage(**data)

Bases: FileTransferMessage

Message for requesting a file upload

Contains a unique file_id and the MIME content type. May contain name, size, destination directory and an associated terminal id.

Parameters:

data (Any)

content_type: str
destination_dir: str | None
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'content_type': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'destination_dir': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'file_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'name': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'size': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'terminal_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'type': FieldInfo(annotation=str, required=False, default='file_upload_request')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

name: str | None
size: int | None
terminal_id: str | None
type: str
class opsicommon.messagebus.message.FileUploadResponseMessage(**data)

Bases: FileTransferMessage

Message to send as response to a file upload request

Contains the local path of the file to be uploaded.

Parameters:

data (Any)

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'file_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'path': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=False, default='file_upload_response')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

path: str | None
type: str
class opsicommon.messagebus.message.FileUploadResultMessage(**data)

Bases: FileTransferMessage

Message to send after file upload concluded

Contains the path of the uploaded file.

Parameters:

data (Any)

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'file_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'path': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=False, default='file_upload_result')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

path: str | None
type: str
class opsicommon.messagebus.message.GeneralErrorMessage(**data)

Bases: Message

Base Class for Error Messages

Used to transport Error object via messagebus

Parameters:

data (Any)

error: Error
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'error': FieldInfo(annotation=Error, required=True), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=False, default='general_error')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str
class opsicommon.messagebus.message.JSONRPCRequestMessage(**data)

Bases: Message

Message for triggering an rpc

Requests the execution of an rpc with given parameters on receiving end.

Parameters:

data (Any)

api_version: str
method: str
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'api_version': FieldInfo(annotation=str, required=False, default='1'), 'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'method': FieldInfo(annotation=str, required=True), 'params': FieldInfo(annotation=tuple[Any, ...], required=False, default=()), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'rpc_id': FieldInfo(annotation=Union[str, int], required=False, default_factory=<lambda>), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=False, default='jsonrpc_request')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

params: tuple[Any, ...]
rpc_id: str | int
type: str
class opsicommon.messagebus.message.JSONRPCResponseMessage(**data)

Bases: Message

Message for transmitting result of rpc

Is response to JSONRPCRequestMessage and contains either result or an error. rpc_id matches the one specified in the corresponding JSONRPCRequestMessage.

Parameters:

data (Any)

error: Any
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'error': FieldInfo(annotation=Any, required=False, default=None), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'result': FieldInfo(annotation=Any, required=False, default=None), 'rpc_id': FieldInfo(annotation=Union[str, int], required=True), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=False, default='jsonrpc_response')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

result: Any
rpc_id: str | int
type: str
class opsicommon.messagebus.message.Message(**data)

Bases: BaseModel, ABC

Parameters:

data (Any)

back_channel: str | None
channel: str
created: int
expires: int
classmethod from_dict(data)
Parameters:

data (dict[str, Any])

Return type:

TypeVar(MessageT, bound= Message)

classmethod from_msgpack(data)
Parameters:

data (bytes)

Return type:

TypeVar(MessageT, bound= Message)

id: UUID4Str
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

ref_id: str | None
property response_channel: str
sender: str
to_dict(none_values=False)
Parameters:

none_values (bool)

Return type:

dict[str, Any]

to_msgpack(none_values=False)
Parameters:

none_values (bool)

Return type:

bytes

type: str
opsicommon.messagebus.message.MessageErrorEnum

alias of ErrorCode

class opsicommon.messagebus.message.MessageType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: StrEnum

CHANNEL_SUBSCRIPTION_EVENT = 'channel_subscription_event'
CHANNEL_SUBSCRIPTION_REQUEST = 'channel_subscription_request'
EVENT = 'event'
FILE_CHUNK = 'file_chunk'
FILE_DOWNLOAD_INFORMATION = 'file_download_information'
FILE_DOWNLOAD_REQUEST = 'file_download_request'
FILE_TRANSFER_ERROR = 'file_transfer_error'
FILE_UPLOAD_REQUEST = 'file_upload_request'
FILE_UPLOAD_RESPONSE = 'file_upload_response'
FILE_UPLOAD_RESULT = 'file_upload_result'
GENERAL_ERROR = 'general_error'
JSONRPC_REQUEST = 'jsonrpc_request'
JSONRPC_RESPONSE = 'jsonrpc_response'
PROCESS_DATA_READ = 'process_data_read'
PROCESS_DATA_WRITE = 'process_data_write'
PROCESS_ERROR = 'process_error'
PROCESS_START_EVENT = 'process_start_event'
PROCESS_START_REQUEST = 'process_start_request'
PROCESS_STOP_EVENT = 'process_stop_event'
PROCESS_STOP_REQUEST = 'process_stop_request'
TERMINAL_CLOSE_EVENT = 'terminal_close_event'
TERMINAL_CLOSE_REQUEST = 'terminal_close_request'
TERMINAL_DATA_READ = 'terminal_data_read'
TERMINAL_DATA_WRITE = 'terminal_data_write'
TERMINAL_ERROR = 'terminal_error'
TERMINAL_OPEN_EVENT = 'terminal_open_event'
TERMINAL_OPEN_REQUEST = 'terminal_open_request'
TERMINAL_RESIZE_EVENT = 'terminal_resize_event'
TERMINAL_RESIZE_REQUEST = 'terminal_resize_request'
TRACE_REQUEST = 'trace_request'
TRACE_RESPONSE = 'trace_response'
class opsicommon.messagebus.message.ProcessDataReadMessage(**data)

Bases: ProcessMessage

Message transmitting process output data

Process stdout and stderr output data is contained as bytes.

Parameters:

data (Any)

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'process_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'stderr': FieldInfo(annotation=bytes, required=False, default=b''), 'stdout': FieldInfo(annotation=bytes, required=False, default=b''), 'type': FieldInfo(annotation=str, required=False, default='process_data_read')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

stderr: bytes
stdout: bytes
type: str
class opsicommon.messagebus.message.ProcessDataWriteMessage(**data)

Bases: ProcessMessage

Message transmitting process input data

Process input data (stdin) is contained as bytes.

Parameters:

data (Any)

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'process_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'stdin': FieldInfo(annotation=bytes, required=False, default=b''), 'type': FieldInfo(annotation=str, required=False, default='process_data_write')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

stdin: bytes
type: str
class opsicommon.messagebus.message.ProcessErrorMessage(**data)

Bases: ProcessMessage

Message reporting a process related Error

Used to transport Error object via messagebus.

Parameters:

data (Any)

error: Error
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'error': FieldInfo(annotation=Error, required=True), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'process_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=False, default='process_error')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str
class opsicommon.messagebus.message.ProcessMessage(**data)

Bases: Message, ABC

Message regarding a process.

Contains a unique process_id.

Parameters:

data (Any)

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'process_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

process_id: UUID4Str
class opsicommon.messagebus.message.ProcessStartEventMessage(**data)

Bases: ProcessMessage

Message to respond to ProcessStartRequestMessage

Contains the local process id.

Parameters:

data (Any)

locale_encoding: str | None
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'locale_encoding': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'os_process_id': FieldInfo(annotation=int, required=True, alias_priority=2, validation_alias=AliasChoices(choices=['os_process_id', 'local_process_id'])), 'process_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=False, default='process_start_event')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

os_process_id: int
type: str
class opsicommon.messagebus.message.ProcessStartRequestMessage(**data)

Bases: ProcessMessage

Message requesting to start a process

Contains a unique process_id and the command to execute as tuple. Optional timeout.

Parameters:

data (Any)

command: tuple[str, ...]
env: dict[str, str]
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'command': FieldInfo(annotation=tuple[str, ...], required=False, default=()), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'env': FieldInfo(annotation=dict[str, str], required=False, default_factory=dict), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'process_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'shell': FieldInfo(annotation=bool, required=False, default=False), 'timeout': FieldInfo(annotation=int, required=False, default=0), 'type': FieldInfo(annotation=str, required=False, default='process_start_request')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

shell: bool
timeout: int
type: str
class opsicommon.messagebus.message.ProcessStopEventMessage(**data)

Bases: ProcessMessage

Message to respond to ProcessStopRequestMessage

Contains the exit code of the process. May contain error.

Parameters:

data (Any)

exit_code: int
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'exit_code': FieldInfo(annotation=int, required=True), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'process_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=False, default='process_stop_event')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str
class opsicommon.messagebus.message.ProcessStopRequestMessage(**data)

Bases: ProcessMessage

Message requesting to stop a running process

Contains the local process id.

Parameters:

data (Any)

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'process_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=False, default='process_stop_request')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str
class opsicommon.messagebus.message.TerminalCloseEventMessage(**data)

Bases: TerminalMessage

Message to respond to TerminalCloseRequestMessage

Parameters:

data (Any)

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'terminal_id': FieldInfo(annotation=str, required=True, metadata=[MinLen(min_length=1)]), 'type': FieldInfo(annotation=str, required=False, default='terminal_close_event')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str
class opsicommon.messagebus.message.TerminalCloseRequestMessage(**data)

Bases: TerminalMessage

Message to request a terminal to be closed

Contains terminal_id for open termial to be closed.

Parameters:

data (Any)

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'terminal_id': FieldInfo(annotation=str, required=True, metadata=[MinLen(min_length=1)]), 'type': FieldInfo(annotation=str, required=False, default='terminal_close_request')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str
class opsicommon.messagebus.message.TerminalDataReadMessage(**data)

Bases: TerminalMessage

Message transmitting terminal output data

Terminal output data is contained as bytes.

Parameters:

data (Any)

data: bytes
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'data': FieldInfo(annotation=bytes, required=True), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'terminal_id': FieldInfo(annotation=str, required=True, metadata=[MinLen(min_length=1)]), 'type': FieldInfo(annotation=str, required=False, default='terminal_data_read')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str
class opsicommon.messagebus.message.TerminalDataWriteMessage(**data)

Bases: TerminalMessage

Message transmitting terminal input data

Terminal input data (stdin) is contained as bytes.

Parameters:

data (Any)

data: bytes
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'data': FieldInfo(annotation=bytes, required=True), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'terminal_id': FieldInfo(annotation=str, required=True, metadata=[MinLen(min_length=1)]), 'type': FieldInfo(annotation=str, required=False, default='terminal_data_write')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str
class opsicommon.messagebus.message.TerminalErrorMessage(**data)

Bases: TerminalMessage

Message reporting a terminal related Error

Used to transport Error object via messagebus.

Parameters:

data (Any)

error: Error
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'error': FieldInfo(annotation=Error, required=True), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'terminal_id': FieldInfo(annotation=str, required=True, metadata=[MinLen(min_length=1)]), 'type': FieldInfo(annotation=str, required=False, default='terminal_error')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str
class opsicommon.messagebus.message.TerminalMessage(**data)

Bases: Message, ABC

Message interacting with a terminal

terminal_id is used as an identifier.

Parameters:

data (Any)

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'terminal_id': FieldInfo(annotation=str, required=True, metadata=[MinLen(min_length=1)]), 'type': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

terminal_id: str
class opsicommon.messagebus.message.TerminalOpenEventMessage(**data)

Bases: TerminalMessage

Message to respond to TerminalOpenRequestMessage

Contains number of rows and columns.

Parameters:

data (Any)

cols: int
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'cols': FieldInfo(annotation=int, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'rows': FieldInfo(annotation=int, required=True), 'sender': FieldInfo(annotation=str, required=True), 'terminal_id': FieldInfo(annotation=str, required=True, metadata=[MinLen(min_length=1)]), 'type': FieldInfo(annotation=str, required=False, default='terminal_open_event')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

rows: int
type: str
class opsicommon.messagebus.message.TerminalOpenRequestMessage(**data)

Bases: TerminalMessage

Message requesting to open a terminal

Shell and number of rows and columns can be specified. If a terminal with that id already exists, access to this terminal may be granted resulting in shared access to it.

Parameters:

data (Any)

cols: int | None
env: dict[str, str]
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'cols': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'env': FieldInfo(annotation=dict[str, str], required=False, default_factory=dict), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'rows': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'shell': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'terminal_id': FieldInfo(annotation=str, required=True, metadata=[MinLen(min_length=1)]), 'type': FieldInfo(annotation=str, required=False, default='terminal_open_request')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

rows: int | None
shell: str | None
type: str
class opsicommon.messagebus.message.TerminalResizeEventMessage(**data)

Bases: TerminalMessage

Message to respond to TerminalResizeRequestMessage

Contains new number of rows and columns.

Parameters:

data (Any)

cols: int
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'cols': FieldInfo(annotation=int, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'rows': FieldInfo(annotation=int, required=True), 'sender': FieldInfo(annotation=str, required=True), 'terminal_id': FieldInfo(annotation=str, required=True, metadata=[MinLen(min_length=1)]), 'type': FieldInfo(annotation=str, required=False, default='terminal_resize_event')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

rows: int
type: str
class opsicommon.messagebus.message.TerminalResizeRequestMessage(**data)

Bases: TerminalMessage

Message requesting to resize an open terminal

Contains new number of rows and columns for an already open terminal.

Parameters:

data (Any)

cols: int
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'cols': FieldInfo(annotation=int, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'rows': FieldInfo(annotation=int, required=True), 'sender': FieldInfo(annotation=str, required=True), 'terminal_id': FieldInfo(annotation=str, required=True, metadata=[MinLen(min_length=1)]), 'type': FieldInfo(annotation=str, required=False, default='terminal_resize_request')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

rows: int
type: str
class opsicommon.messagebus.message.TraceRequestMessage(**data)

Bases: Message

Message for tracing transmission times

It contains trace data (timestamp of sending).

Parameters:

data (Any)

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'payload': FieldInfo(annotation=Union[bytes, NoneType], required=False, default=None), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sender': FieldInfo(annotation=str, required=True), 'trace': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'type': FieldInfo(annotation=str, required=False, default='trace_request')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

payload: bytes | None
trace: dict[str, Any]
type: str
class opsicommon.messagebus.message.TraceResponseMessage(**data)

Bases: Message

Message for tracing transmission times (response)

It contains trace data (timestamp of sending, receiving request and response).

Parameters:

data (Any)

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'back_channel': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'channel': FieldInfo(annotation=str, required=True), 'created': FieldInfo(annotation=int, required=False, default_factory=timestamp), 'expires': FieldInfo(annotation=int, required=False, default_factory=<lambda>), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, metadata=[StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern='^[0-9a-f]{8}\\-[0-9a-f]{4}\\-4[0-9a-f]{3}\\-[89ab][0-9a-f]{3}\\-[0-9a-f]{12}$')]), 'payload': FieldInfo(annotation=Union[bytes, NoneType], required=False, default=None), 'ref_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'req_trace': FieldInfo(annotation=dict[str, Any], required=True), 'sender': FieldInfo(annotation=str, required=True), 'trace': FieldInfo(annotation=dict[str, Any], required=True), 'type': FieldInfo(annotation=str, required=False, default='trace_response')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

payload: bytes | None
req_trace: dict[str, Any]
trace: dict[str, Any]
type: str
opsicommon.messagebus.message.timestamp(add_seconds=0.0)

Returns the current time (UTC) as messagebus timestamp. add_seconds can be used to add or subtract seconds from the current time.

Parameters:

add_seconds (float)

Return type:

int

opsicommon.messagebus.process module

This file is part of opsi - https://www.opsi.org

class opsicommon.messagebus.process.Process(process_start_request, send_message, sender='@', back_channel=None)

Bases: object

Parameters:
async close_stdin()
Return type:

None

max_data_size = 8192
async start()
Return type:

None

async stop()
Return type:

None

async write_stdin(data)
Parameters:

data (bytes)

Return type:

None

opsicommon.messagebus.process.get_locale_encoding(shell=False)
Parameters:

shell (bool)

Return type:

str

async opsicommon.messagebus.process.process_messagebus_message(message, send_message, *, sender='@', back_channel=None)
Parameters:

  • message (ProcessMessage)

  • send_message (Callable)

  • sender (str)

  • back_channel (str | None)

Return type:

None

opsicommon.messagebus.process.remove_process(process_id)
Parameters:

process_id (str)

Return type:

None

async opsicommon.messagebus.process.stop_running_processes()
Return type:

None

opsicommon.messagebus.terminal module

This file is part of opsi - https://www.opsi.org

class opsicommon.messagebus.terminal.Terminal(terminal_open_request, send_message, sender='@', back_channel=None, default_shell=None)

Bases: object

Parameters:

  • terminal_open_request (TerminalOpenRequestMessage)

  • send_message (Callable)

  • sender (str)

  • back_channel (str | None)

  • default_shell (str | None)

async close(message=None, use_terminals_lock=True)
Parameters:
Return type:

None

default_cols = 120
default_rows = 30
get_cwd()
Return type:

Path | None

idle_timeout = 28800
max_cols = 300
max_rows = 100
async process_message(message)
Parameters:

message (TerminalDataWriteMessage | TerminalResizeRequestMessage | TerminalCloseRequestMessage)

Return type:

None

pty_reader_block_size = 16384
async reuse(terminal_open_request)
Parameters:

terminal_open_request (TerminalOpenRequestMessage)

Return type:

None

async set_size(rows=None, cols=None)
Parameters:

  • rows (int | None)

  • cols (int | None)

Return type:

None

async start()
Return type:

None

property terminal_id: str
async opsicommon.messagebus.terminal.process_messagebus_message(message, send_message, *, sender='@', back_channel=None)
Parameters:

  • message (TerminalMessage)

  • send_message (Callable)

  • sender (str)

  • back_channel (str | None)

Return type:

None

opsicommon.messagebus.terminal.remove_terminal(terminal_id)
Parameters:

terminal_id (str)

Return type:

None

opsicommon.messagebus.terminal.start_pty(shell, rows=30, cols=120, cwd=None, env=None)
Parameters:

  • shell (str)

  • rows (int | None)

  • cols (int | None)

  • cwd (str | None)

  • env (dict[str, str] | None)

Return type:

tuple[int, Callable, Callable, Callable, Callable]

async opsicommon.messagebus.terminal.stop_running_terminals()
Return type:

None

Module contents

This file is part of opsi - https://www.opsi.org