dasf.messaging.constants module

Enums within the DASF Messaging Framework.

Classes:

`MessageType`(value)	Supported message types.
`ModuleRequest`(*, payload, context, properties)	A request to send via the message broker.
`ModuleResponse`(*[, messageId, publishTime])	A request to send via the message broker.
`PropertyKeys`(value)	Property keys for a message to the message broker.
`RequestProperties`(*, messageType, ...)	A DASF request for a consumer.
`RequestPropertiesBase`(*, messageType, ...)	Base class for a DASF request or response.
`ResponseProperties`(*, messageType, ...[, ...])	A DASF response for a producer.
`Status`(value)	Status flag of a request.

Functions:

generate_enum_doc(class_)

Utility function to generate the docstring for an enum.

class dasf.messaging.constants.MessageType(value)[source]

Bases: str, Enum

Supported message types.

The following values are valid:

PING ("ping")
PONG ("pong")
REQUEST ("request")
RESPONSE ("response")
LOG ("log")
INFO ("info")
PROGRESS ("progress")
API_INFO ("api_info")

Attributes:

`API_INFO`
`INFO`
`LOG`
`PING`
`PONG`
`PROGRESS`
`REQUEST`
`RESPONSE`

API_INFO = 'api_info'

INFO = 'info'

LOG = 'log'

PING = 'ping'

PONG = 'pong'

PROGRESS = 'progress'

REQUEST = 'request'

RESPONSE = 'response'

class dasf.messaging.constants.ModuleRequest(*, payload: Annotated[str, EncodedStr(encoder=Base64Encoder)], context: str, properties: RequestProperties)[source]

Bases: BaseModel

A request to send via the message broker.

Attributes:

`context`
`model_config`	Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
`payload`
`properties`

context: str

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

payload: Annotated[str, EncodedStr(encoder=Base64Encoder)]

properties: RequestProperties

class dasf.messaging.constants.ModuleResponse(*, messageId: str = '', payload: Annotated[str, EncodedStr(encoder=Base64Encoder)], publishTime: str = '', properties: ResponseProperties)[source]

Bases: BaseModel

A request to send via the message broker.

Attributes:

`messageId`
`model_config`	Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
`payload`
`properties`
`publishTime`

messageId: str

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

payload: Annotated[str, EncodedStr(encoder=Base64Encoder)]

properties: ResponseProperties

publishTime: str

class dasf.messaging.constants.PropertyKeys(value)[source]

Bases: str, Enum

Property keys for a message to the message broker.

The following values are valid:

REQUEST_CONTEXT ("requestContext")
RESPONSE_TOPIC ("response_topic")
SOURCE_TOPIC ("source_topic")
REQUEST_MESSAGEID ("requestMessageId")
MESSAGE_TYPE ("messageType")
FRAGMENT ("fragment")
NUM_FRAGMENTS ("num_fragments")
STATUS ("status")

Attributes:

`FRAGMENT`
`MESSAGE_TYPE`
`NUM_FRAGMENTS`
`REQUEST_CONTEXT`
`REQUEST_MESSAGEID`
`RESPONSE_TOPIC`
`SOURCE_TOPIC`
`STATUS`

FRAGMENT = 'fragment'

MESSAGE_TYPE = 'messageType'

NUM_FRAGMENTS = 'num_fragments'

REQUEST_CONTEXT = 'requestContext'

REQUEST_MESSAGEID = 'requestMessageId'

RESPONSE_TOPIC = 'response_topic'

SOURCE_TOPIC = 'source_topic'

STATUS = 'status'

class dasf.messaging.constants.RequestProperties(*, messageType: MessageType, requestContext: int | Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None, ascii_only=None)], response_topic: Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None, ascii_only=None)])[source]

Bases: RequestPropertiesBase

A DASF request for a consumer.

Attributes:

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

messageType: MessageType

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

requestContext: int | Annotated[str, StringConstraints(min_length=1)]

response_topic: Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None, ascii_only=None)]

class dasf.messaging.constants.RequestPropertiesBase(*, messageType: MessageType, requestContext: int | Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None, ascii_only=None)])[source]

Bases: BaseModel

Base class for a DASF request or response.

Attributes:

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

messageType: MessageType

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

requestContext: int | Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None, ascii_only=None)]

class dasf.messaging.constants.ResponseProperties(*, messageType: MessageType, requestContext: int | Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None, ascii_only=None)], requestMessageId: str | None = None, source_topic: str | None = None, fragment: int | None = None, num_fragments: int | None = None, status: Status | None = None, info: str | None = None, api_info: str | None = None, response_topic: Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None, ascii_only=None)] | None = None)[source]

Bases: RequestPropertiesBase

A DASF response for a producer.

Attributes:

`api_info`
`fragment`
`info`
`model_config`	Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
`num_fragments`
`requestMessageId`
`response_topic`
`source_topic`
`status`

api_info: str | None

fragment: int | None

info: str | None

messageType: MessageType

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

num_fragments: int | None

requestContext: int | Annotated[str, StringConstraints(min_length=1)]

requestMessageId: str | None

response_topic: Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None, ascii_only=None)] | None

source_topic: str | None

status: Status | None

class dasf.messaging.constants.Status(value)[source]

Bases: str, Enum

Status flag of a request.

The following values are valid:

SUCCESS ("success")
ERROR ("error")
RUNNING ("running")

Attributes:

`ERROR`
`RUNNING`
`SUCCESS`

ERROR = 'error'

RUNNING = 'running'

SUCCESS = 'success'

dasf.messaging.constants.generate_enum_doc(class_: Type[Enum]) → Type[Enum][source]: Utility function to generate the docstring for an enum.