`multicloud.pubsub`¶

Cloud-agnostic publish/subscribe messaging. See the Pub/Sub guide for task-focused walkthroughs.

Cloud-agnostic pubsub client package.

The user-facing layer: TopicClient and SubscriptionClient plus the value objects callers pass to their operations. Provider-specific code lives in sibling packages (pubsub_aws, pubsub_gcp, pubsub_ali) and is hidden behind the registry.

SubscriptionClient ¶

SubscriptionClient(
    *,
    provider: str,
    subscription_name: str,
    region: Optional[str] = None,
    endpoint: Optional[str] = None,
    nack_visibility_timeout: Optional[timedelta] = None,
    credentials_overrider: Optional[
        CredentialsOverrider
    ] = None,
    **provider_kwargs: Any
)

Cloud-agnostic subscription client for receiving and acknowledging messages.

The user-facing API surface for subscribing. Construction takes a provider id and a unified set of cloud-agnostic options; the underlying provider class is selected from the registry.

Parameters:

Name	Type	Description	Default
`provider`	`str`	Provider id (e.g., `"aws"`, `"gcp"`, `"ali"`). See the registry for all registered providers.	required
`subscription_name`	`str`	Subscription name/identifier.	required
`region`	`Optional[str]`	Region (interpreted per provider). Optional.	`None`
`endpoint`	`Optional[str]`	Endpoint override URL. Optional.	`None`
`nack_visibility_timeout`	`Optional[timedelta]`	Default visibility timeout for nacked messages. Optional; defaults to 0 (immediate redelivery).	`None`
`credentials_overrider`	`Optional[CredentialsOverrider]`	Override default credential resolution. Optional; used for test isolation, cross-account access, or role assumption.	`None`

Note

Credentials are sourced from environment variables or system default credential discovery (environment variables, credential files, or cloud-native authentication like IAM roles or workload identity).

Use credentials_overrider for structured credential overrides (session credentials, role assumption, web identity). Individual credential parameters (aws_access_key_id, etc.) are not supported.

receive ¶

receive() -> Message

Receive a single message from the subscription (blocking).

Returns:

Type	Description
`Message`	The received message with ack_id populated.

send_ack ¶

send_ack(ack_id: AckID) -> None

Acknowledge a single message.

Parameters:

Name	Type	Description	Default
`ack_id`	`AckID`	The acknowledgment ID from the received message.	required

Raises:

Type	Description
`InvalidArgumentException`	if ack_id is None

send_acks ¶

send_acks(ack_ids: List[AckID]) -> Future[None]

Acknowledge multiple messages in a batch.

Parameters:

Name	Type	Description	Default
`ack_ids`	`List[AckID]`	List of acknowledgment IDs.	required

Returns:

Type	Description
`Future[None]`	A Future that completes when all acks are processed.

Raises:

Type	Description
`InvalidArgumentException`	if any ack_id is None

send_nack ¶

send_nack(
    ack_id: AckID,
    *,
    visibility_timeout: Optional[timedelta] = None
) -> None

Negatively acknowledge a single message (requeue for redelivery).

Parameters:

Name	Type	Description	Default
`ack_id`	`AckID`	The acknowledgment ID from the received message.	required
`visibility_timeout`	`Optional[timedelta]`	Optional visibility timeout for the nacked message.	`None`

Raises:

Type	Description
`InvalidArgumentException`	if ack_id is None or timeout is negative

send_nacks ¶

send_nacks(
    ack_ids: List[AckID],
    *,
    visibility_timeout: Optional[timedelta] = None
) -> Future[None]

Negatively acknowledge multiple messages in a batch.

Parameters:

Name	Type	Description	Default
`ack_ids`	`List[AckID]`	List of acknowledgment IDs.	required
`visibility_timeout`	`Optional[timedelta]`	Optional visibility timeout for all nacked messages.	`None`

Returns:

Type	Description
`Future[None]`	A Future that completes when all nacks are processed.

Raises:

Type	Description
`InvalidArgumentException`	if ack_ids is None, any ack_id is None, or timeout is negative

can_nack ¶

can_nack() -> bool

Check if the provider supports negative acknowledgment.

Returns:

Type	Description
`bool`	True if nack is supported, False otherwise.

get_attributes ¶

get_attributes() -> GetAttributeResult

Retrieve subscription attributes.

Returns:

Type	Description
`GetAttributeResult`	Subscription attributes including approximate message count.

is_retryable ¶

is_retryable(error: BaseException) -> bool

Check if an error is retryable.

Parameters:

Name	Type	Description	Default
`error`	`BaseException`	The error to check.	required

Returns:

Type	Description
`bool`	True if the error is transient and retryable, False otherwise.

close ¶

close() -> None

Flush pending acknowledgments and close the subscription.

TopicClient ¶

TopicClient(
    *,
    provider: str,
    topic_name: str,
    region: Optional[str] = None,
    endpoint: Optional[str] = None,
    credentials_overrider: Optional[
        CredentialsOverrider
    ] = None,
    **provider_kwargs: Any
)

Cloud-agnostic topic client for publishing messages.

The user-facing API surface for publishing. Construction takes a provider id and a unified set of cloud-agnostic options; the underlying provider class is selected from the registry. Operations (send) take cloud-agnostic value objects and never reference a provider.

Parameters:

Name	Type	Description	Default
`provider`	`str`	Provider id (e.g., `"awssns"`, `"awssqs"`, `"gcp"`, `"ali"`). See the registry for all registered providers.	required
`topic_name`	`str`	Topic name/identifier.	required
`region`	`Optional[str]`	Region identifier (interpretation varies by provider). Optional; falls back to provider default credential discovery.	`None`
`endpoint`	`Optional[str]`	Endpoint override URL. Optional.	`None`
`credentials_overrider`	`Optional[CredentialsOverrider]`	Override default credential resolution. Optional; used for test isolation, cross-account access, or role assumption.	`None`
`**provider_kwargs`	`Any`	Additional provider-specific keyword arguments. See provider implementation documentation for available options.	`{}`

Note

Credentials are sourced from environment variables or system default credential discovery (environment variables, credential files, or cloud-native authentication like IAM roles or workload identity).

Use credentials_overrider for structured credential overrides (session credentials, role assumption, web identity). Individual credential parameters (aws_access_key_id, etc.) are not supported.

send ¶

send(message: Message) -> None

Send a single message to the topic.

Pubsub send() accepts a Message object because messages can carry metadata and lifecycle hooks that are naturally composed with the builder pattern.

Parameters:

Name	Type	Description	Default
`message`	`Message`	The message to send.	required

Raises:

Type	Description
`InvalidArgumentException`	if message is invalid

close ¶

close() -> None

Flush pending messages and close the topic.

AckID ¶

Bases: Protocol

Marker protocol for provider-specific acknowledgment IDs.

GetAttributeResult `dataclass` ¶

GetAttributeResult(
    name: Optional[str] = None,
    topic: Optional[str] = None,
    approximate_number_of_messages: Optional[int] = None,
)

Result of getting subscription attributes.

Attributes:

Name	Type	Description
`name`	`Optional[str]`	Subscription name or identifier (provider-specific format)
`topic`	`Optional[str]`	Topic name or resource identifier (provider-specific format)
`approximate_number_of_messages`	`Optional[int]`	Approximate count of messages in subscription (optional)

Message `dataclass` ¶

Message(
    loggable_id: Optional[str] = None,
    body: Optional[bytes] = None,
    metadata: Dict[str, str] = dict(),
    ack_id: Optional[AckID] = None,
    as_func: Optional[AsFunction] = None,
    before_send: Optional[SendHook] = None,
    after_send: Optional[SendHook] = None,
)

Message represents a message in the pubsub system.

Messages contain the actual data being transmitted along with metadata and acknowledgment information when received from subscriptions.

Message is mutable by design because it is assembled incrementally through the builder and may be enriched during pubsub flow (for example: metadata, acknowledgement id, loggable id, and send hooks). Use Message.builder() to create new messages.

Attributes:

Name	Type	Description
`loggable_id`	`Optional[str]`	Loggable ID for this message (useful for debugging). This is set for received messages. Should not be set for outgoing messages.
`body`	`Optional[bytes]`	The message body content as bytes.
`metadata`	`Dict[str, str]`	A map of metadata key-value pairs.
`ack_id`	`Optional[AckID]`	The acknowledgment ID for this message. Only set for received messages.
`as_func`	`Optional[AsFunction]`	Optional function for driver-specific type conversion.
`before_send`	`Optional[SendHook]`	Optional callback invoked before sending.
`after_send`	`Optional[SendHook]`	Optional callback invoked after sending.

Builder ¶

Builder()

Builder for creating Message instances.

Example

message = (Message.builder() ... .with_body("Hello World") ... .with_metadata("key", "value") ... .build())

with_body ¶

with_body(body: object) -> 'Message.Builder'

Set the message body.

Parameters:

Name	Type	Description	Default
`body`	`object`	The message body as bytes or string (will be UTF-8 encoded)	required

Returns:

Type	Description
`'Message.Builder'`	This builder instance for chaining

Raises:

Type	Description
`TypeError`	if body is not bytes or str

with_metadata ¶

with_metadata(
    metadata: object, value: Optional[str] = None
) -> "Message.Builder"

Set the message metadata.

Can be called with a dict to set all metadata, or with key-value pairs to add individual metadata entries.

Parameters:

Name	Type	Description	Default
`metadata`	`object`	Either a dict of metadata key-value pairs, or a string key	required
`value`	`Optional[str]`	If metadata is a string key, this is the value. Defaults to empty string.	`None`

Returns:

Type	Description
`'Message.Builder'`	This builder instance for chaining

Raises:

Type	Description
`TypeError`	if metadata is neither dict nor str

with_loggable_id ¶

with_loggable_id(loggable_id: str) -> 'Message.Builder'

Set the loggable ID (typically used internally by drivers).

Parameters:

Name	Type	Description	Default
`loggable_id`	`str`	The loggable ID	required

Returns:

Type	Description
`'Message.Builder'`	This builder instance for chaining

with_ack_id ¶

with_ack_id(ack_id: AckID) -> 'Message.Builder'

Set the acknowledgment ID (typically used internally by drivers).

Parameters:

Name	Type	Description	Default
`ack_id`	`AckID`	The acknowledgment ID	required

Returns:

Type	Description
`'Message.Builder'`	This builder instance for chaining

with_as_func ¶

with_as_func(as_func: AsFunction) -> 'Message.Builder'

Set the as function for driver-specific type conversion.

Parameters:

Name	Type	Description	Default
`as_func`	`AsFunction`	The as function	required

Returns:

Type	Description
`'Message.Builder'`	This builder instance for chaining

with_before_send ¶

with_before_send(
    before_send: SendHook,
) -> "Message.Builder"

Set the before send callback.

Parameters:

Name	Type	Description	Default
`before_send`	`SendHook`	The before send callback	required

Returns:

Type	Description
`'Message.Builder'`	This builder instance for chaining

with_after_send ¶

with_after_send(after_send: SendHook) -> 'Message.Builder'

Set the after send callback.

Parameters:

Name	Type	Description	Default
`after_send`	`SendHook`	The after send callback	required

Returns:

Type	Description
`'Message.Builder'`	This builder instance for chaining

build ¶

build() -> 'Message'

Build and return the configured Message instance.

Returns:

Type	Description
`'Message'`	The Message with the configured properties

get_byte_size ¶

get_byte_size() -> int

Return the encoded size used for per-message and batch limits.

Includes message body bytes plus metadata key/value UTF-8 byte lengths.

as_type ¶

as_type(target: object) -> bool

Convert the message to a driver-specific type.

Parameters:

Name	Type	Description	Default
`target`	`object`	The target type to convert to	required

Returns:

Type	Description
`bool`	True if the conversion was successful, False otherwise

builder `staticmethod` ¶

builder() -> 'Message.Builder'

Create a new message builder.

Returns:

Type	Description
`'Message.Builder'`	A new Builder instance

multicloud.pubsub¶

SubscriptionClient ¶

receive ¶

send_ack ¶

send_acks ¶

send_nack ¶

send_nacks ¶

can_nack ¶

get_attributes ¶

is_retryable ¶

close ¶

TopicClient ¶

send ¶

close ¶

AckID ¶

GetAttributeResult dataclass ¶

Message dataclass ¶

Builder ¶

with_body ¶

with_metadata ¶

with_loggable_id ¶

with_ack_id ¶

with_as_func ¶

with_before_send ¶

with_after_send ¶

build ¶

get_byte_size ¶

as_type ¶

builder staticmethod ¶

`multicloud.pubsub`¶

GetAttributeResult `dataclass` ¶

Message `dataclass` ¶

builder `staticmethod` ¶