saturnin.base.types

Saturnin common type definitions and constants

This module contains:

1. Type aliases and new types for type annotations.

2. Commonly used constants like platform, vendor and mime type identifiers.

3. Exceptions.

4. Sentinels.

5. Enums.

6. Dataclasses.

Types for annotations

saturnin.base.types.TSupplement: TypeAlias = dict[str, typing.Any] | None

name/value dictionary

saturnin.base.types.Token = saturnin.base.types.Token

Message token

saturnin.base.types.RoutingID = saturnin.base.types.RoutingID

Routing ID

Constants

saturnin.base.types.PLATFORM_OID: Final[str] = '1.3.6.1.4.1.53446.1.1.0'

Platform OID (firebird.butler.platform.saturnin)

saturnin.base.types.PLATFORM_UID: Final[UUID] = UUID('46cd9e8a-c697-5cb5-abb5-bceac5a17075')

Platform UID (uuid5() - NAMESPACE_OID)

saturnin.base.types.PLATFORM_VERSION: Final[str] = '0.8.0'

Platform version (semver)

saturnin.base.types.VENDOR_OID: Final[str] = '1.3.6.1.4.1.53446.1.2.0'

Platform vendor OID (firebird.butler.vendor.firebird)

saturnin.base.types.VENDOR_UID: Final[UUID] = UUID('c26a6600-d579-5ec7-a9d6-5b0a8b214d3f')

Platform vendor UID (uuid5() - NAMESPACE_OID)

saturnin.base.types.MIME_TYPE_PROTO: Final[MIME] = MIME('application/x.fb.proto')

MIME type for protobuf messages

saturnin.base.types.MIME_TYPE_TEXT: Final[MIME] = MIME('text/plain')

MIME type for plain text

saturnin.base.types.MIME_TYPE_BINARY: Final[MIME] = MIME('application/octet-stream')

MIME type for binary data

saturnin.base.types.SECTION_LOCAL_ADDRESS: Final[str] = 'local_address'

Configuration section name for local service addresses

saturnin.base.types.SECTION_NODE_ADDRESS: Final[str] = 'node_address'

Configuration section name for node service addresses

saturnin.base.types.SECTION_NET_ADDRESS: Final[str] = 'net_address'

Configuration section name for network service addresses

saturnin.base.types.SECTION_SERVICE_UID: Final[str] = 'service_uid'

Configuration section name for service UIDs

saturnin.base.types.SECTION_PEER_UID: Final[str] = 'peer_uid'

Configuration section name for service peer UIDs

saturnin.base.types.SECTION_BUNDLE: Final[str] = 'bundle'

Default configuration section name for service bundle

saturnin.base.types.SECTION_SERVICE: Final[str] = 'service'

Default configuration section name for single service

saturnin.base.types.PROTO_PEER: Final[str] = 'firebird.butler.PeerIdentification'

protobuf ID for peer information message

Exceptions

exception saturnin.base.types.InvalidMessageError(*args, **kwargs)

Bases: Error

A formal error was detected in a message

exception saturnin.base.types.ChannelError(*args, **kwargs)

Bases: Error

Transmission channel error

exception saturnin.base.types.ServiceError(*args, **kwargs)

Bases: Error

Error raised by service

exception saturnin.base.types.ClientError(*args, **kwargs)

Bases: Error

Error raised by Client

exception saturnin.base.types.StopError(*args, **kwargs)

Bases: Error

Exception that should stop further processing.

exception saturnin.base.types.RestartError(*args, **kwargs)

Bases: Error

Exception signaling that restart is needed for further processing.

Sentinels

saturnin.base.types.INVALID: Final[Sentinel] = INVALID

Sentinel for return values that indicates failed message processing

saturnin.base.types.TIMEOUT: Final[Sentinel] = TIMEOUT

Sentinel for return values that indicates timeout expiration

saturnin.base.types.RESTART: Final[Sentinel] = RESTART

Sentinel for return values that indicates restart request

Enums

class saturnin.base.types.Origin(value)

Bases: IntEnum

Origin of received message in protocol context.

peer_role() → Origin

Returns the peer’s role, which is the logical opposite of this instance’s role.

For example, if this origin is Origin.SERVICE (or Origin.PROVIDER), the peer’s role returned will be Origin.CLIENT (or Origin.CONSUMER). If this origin is Origin.ANY or Origin.UNKNOWN, it returns itself.

Return type:

Origin

ANY = 4

CLIENT = 3

CONSUMER = 3

PROVIDER = 2

SERVICE = 2

UNKNOWN = 1

class saturnin.base.types.SocketMode(value)

Bases: IntEnum

ZeroMQ socket mode.

BIND = 2

CONNECT = 3

UNKNOWN = 1

class saturnin.base.types.Direction(value)

Bases: IntFlag

ZeroMQ socket direction of transmission.

_generate_next_value_(start, count, last_values)

Generate the next value when not given.

name: the name of the member start: the initial start value or None count: the number of existing members last_values: the last value assigned or None

BOTH = 3

IN = 1

NONE = 0

OUT = 2

class saturnin.base.types.SocketType(value)

Bases: IntEnum

ZeroMQ socket type.

DEALER = 5

PAIR = 0

PUB = 1

PULL = 7

PUSH = 8

ROUTER = 6

STREAM = 11

SUB = 2

UNKNOWN_TYPE = -1

XPUB = 9

XSUB = 10

class saturnin.base.types.State(value)

Bases: IntEnum

General state information.

ABORTED = 6

BLOCKED = 3

CREATED = 1

FINISHED = 5

READY = 1

RUNNING = 2

STOPPED = 4

SUSPENDED = 4

TERMINATED = 6

UNKNOWN_STATE = 0

WAITING = 3

class saturnin.base.types.PipeSocket(value)

Bases: IntEnum

Data Pipe Socket identification.

INPUT = 1

OUTPUT = 2

UNKNOWN_PIPE_SOCKET = 0

class saturnin.base.types.FileOpenMode(value)

Bases: IntEnum

File open mode.

APPEND = 4

CREATE = 2

READ = 1

RENAME = 5

UNKNOWN_FILE_OPEN_MODE = 0

WRITE = 3

class saturnin.base.types.Outcome(value)

Bases: Enum

Service execution outcome.

ERROR = 'ERROR'

OK = 'OK'

UNKNOWN = 'UNKNOWN'

class saturnin.base.types.ButlerInterface(value)

Bases: IntEnum

Base class for service API code enumerations (FBSP interfaces).

classmethod get_uid() → UUID

Returns interface UUID.

Return type:

UUID

Dataclasses

class saturnin.base.types.AgentDescriptor(uid: UUID, name: str, version: str, vendor_uid: UUID, classification: str, platform_uid: UUID = UUID('46cd9e8a-c697-5cb5-abb5-bceac5a17075'), platform_version: str = '0.8.0', supplement: TSupplement = None)

Bases: Distinct

Service or Client descriptor dataclass.

Parameters:

• uid (UUID) – Agent ID

• name (str) – Agent name

• version (str) – Agent version string

• vendor_uid (UUID) – Vendor ID

• classification (str) – Agent classification string

• platform_uid (UUID) – Butler platform ID

• platform_version (str) – Butler platform version string

• supplement (TSupplement) – Optional list of supplemental information

copy() → Self

Returns copy of this AgentDescriptor instance.

Return type:

Self

get_key() → UUID

Returns uid (instance key). Used for instance hash computation.

Return type:

UUID

replace(**changes) → Self

Creates a new AgentDescriptor, replacing fields with values from changes.

Return type:

Self

classification: str

name: str

platform_uid: uuid.UUID = UUID('46cd9e8a-c697-5cb5-abb5-bceac5a17075')

platform_version: str = '0.8.0'

supplement: TSupplement = None

uid: uuid.UUID

vendor_uid: uuid.UUID

version: str

class saturnin.base.types.PeerDescriptor(uid: UUID, pid: int, host: str, supplement: TSupplement = None)

Bases: Distinct

Peer descriptor.

Parameters:

• uid (UUID) – Peer ID

• pid (int) – Peer process ID

• host (str) – Host name

• supplement (TSupplement) – Optional list of supplemental information

classmethod from_proto(proto: Any) → Self

Creates new PeerDescriptor from firebird.butler.PeerIdentification protobuf message.

Parameters:

proto (Any)

Return type:

Self

as_proto() → Any

Returns firebird.butler.PeerIdentification protobuf message initialized from instance data.

Return type:

Any

copy() → PeerDescriptor

Returns copy of this PeerDescriptor instance.

Return type:

PeerDescriptor

get_key() → UUID

Returns uid (instance key). Used for instance hash computation.

Return type:

UUID

replace(**changes) → Self

Creates a new PeerDescriptor, replacing fields with values from changes.

Return type:

Self

host: str

pid: int

supplement: TSupplement = None

uid: uuid.UUID

class saturnin.base.types.ServiceDescriptor(agent: AgentDescriptor, api: list[ButlerInterface], description: str, facilities: list[str], factory: str, config: Callable[[], Config])

Bases: Distinct

Service descriptor.

Parameters:

• agent (AgentDescriptor) – Service agent descriptor

• api (list[ButlerInterface]) – Service FBSP API description or None for microservice

• description (str) – Text describing the service

• facilities (list[str]) – List of Saturnin facilities that this service uses

• factory (str) – Locator string for service factory (e.g. like ‘my_package.my_module:my_svc_factory_class’)

• config (Callable[[], Config]) – Service configuration factory (a callable that returns a Config object)

get_key() → UUID

Returns agent.uid (instance key). Used for instance hash computation.

Return type:

UUID

agent: AgentDescriptor

api: list[ButlerInterface]

config: Callable[[], Config]

description: str

facilities: list[str]

factory: str

class saturnin.base.types.ApplicationDescriptor(uid: UUID, name: str, version: str, vendor_uid: UUID, classification: str, description: str, cli_command: str | None = None, recipe_factory: str | None = None)

Bases: Distinct

Application descriptor.

Parameters:

• uid (UUID) – Application ID

• name (str) – Application name

• version (str) – Application version string

• vendor_uid (UUID) – Vendor ID

• classification (str) – Application classification string

• description (str) – Text describing the application

• cli_command (str | None) – Locator string for application typer command (e.g. like ‘my_app.cli:app_cmd’)

• recipe_factory (str | None) – Locator string for application recipe factory (a callable that returns string)

get_key() → UUID

Returns uid (instance key). Used for instance hash computation.

Return type:

UUID

classification: str

cli_command: str | None = None

description: str

name: str

recipe_factory: str | None = None

uid: UUID

vendor_uid: UUID

version: str

class saturnin.base.types.PrioritizedItem(priority: int, item: Any)

Bases: object

Prioritized item for use with heapq to implement priority queue.

Parameters:

• priority (int) – Item priority

• item (Any) – Prioritized item

item: Any

priority: int