Module `spin_sdk.wit.imports.mqtt`

Global variables

var Error: Errors related to interacting with Mqtt

Classes

class Connection

Expand source code

class Connection:
    
    @classmethod
    def open(cls, address: str, username: str, password: str, keep_alive_interval_in_secs: int) -> Self:
        """
        Open a connection to the Mqtt instance at `address`.
        
        Raises: `spin_sdk.wit.types.Err(spin_sdk.wit.imports.mqtt.Error)`
        """
        raise NotImplementedError
    def publish(self, topic: str, payload: bytes, qos: Qos) -> None:
        """
        Publish an Mqtt message to the specified `topic`.
        
        Raises: `spin_sdk.wit.types.Err(spin_sdk.wit.imports.mqtt.Error)`
        """
        raise NotImplementedError
    def __enter__(self) -> Self:
        """Returns self"""
        return self
                                
    def __exit__(self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None) -> bool | None:
        """
        Release this resource.
        """
        raise NotImplementedError

Static methods

def open(address: str, username: str, password: str, keep_alive_interval_in_secs: int) ‑> Self: Open a connection to the Mqtt instance at address.

Raises: Err(Error)

Methods

def publish(self, topic: str, payload: bytes, qos: Qos) ‑> None

Expand source code

def publish(self, topic: str, payload: bytes, qos: Qos) -> None:
    """
    Publish an Mqtt message to the specified `topic`.
    
    Raises: `spin_sdk.wit.types.Err(spin_sdk.wit.imports.mqtt.Error)`
    """
    raise NotImplementedError

Publish an Mqtt message to the specified topic.

Raises: Err(Error)

class Error_ConnectionFailed (value: str)

Expand source code

@dataclass
class Error_ConnectionFailed:
    value: str

Error_ConnectionFailed(value: str)

Instance variables

var value : str

class Error_InvalidAddress

Expand source code

@dataclass
class Error_InvalidAddress:
    pass

Error_InvalidAddress()

class Error_Other (value: str)

Expand source code

@dataclass
class Error_Other:
    value: str

Error_Other(value: str)

Instance variables

var value : str

class Error_TooManyConnections

Expand source code

@dataclass
class Error_TooManyConnections:
    pass

Error_TooManyConnections()

class Qos (*args, **kwds)

Expand source code

class Qos(Enum):
    """
    QoS for publishing Mqtt messages
    """
    AT_MOST_ONCE = 0
    AT_LEAST_ONCE = 1
    EXACTLY_ONCE = 2

QoS for publishing Mqtt messages

Ancestors

enum.Enum

Class variables

var AT_LEAST_ONCE
var AT_MOST_ONCE
var EXACTLY_ONCE