Python API reference

Core classes and functions for extending the MQTT Gateway

This page documents the main Python components you can use when extending or integrating with the gateway.

Constants

Flow action types

The gateway supports two action types for flows, defined in src/constants.py:1-2:

FLOW_ACTION_STORE_DB = "STORE_DB"
FLOW_ACTION_POST_ENDPOINT = "POST_ENDPOINT"

Use these constants when creating or querying flows programmatically to ensure consistency.

Configuration

Settings

Configuration dataclass loaded from environment variables.

Source: src/config.py:8-26

db_host · string · required

Database hostname (from DB_HOST env var)

db_port · integer · default: "3306"

Database port (from DB_PORT env var)

db_name · string · required

Database name (from DB_NAME env var)

db_user · string · required

Database username (from DB_USER env var)

db_password · string · required

Database password (from DB_PASSWORD env var)

log_dir · string · default: "./log"

Log file directory (from LOG_DIR env var)

http_timeout_seconds · integer · default: "10"

Timeout for HTTP POST requests (from HTTP_TIMEOUT_SECONDS env var)

mqtt_client_id · string · default: "mqtt-gateway"

MQTT client identifier (from MQTT_CLIENT_ID env var)

mqtt_keepalive · integer · default: "60"

MQTT keepalive interval in seconds (from MQTT_KEEPALIVE env var)

flows_reload_interval_seconds · integer · default: "600"

How often to reload flows from database (from FLOWS_RELOAD_INTERVAL_SECONDS env var)

Property:

sqlalchemy_url · string

Computed SQLAlchemy connection URL in format: mysql+pymysql://{user}:{password}@{host}:{port}/{name}

load_settings()

Loads configuration from environment variables using python-dotenv.

from src.config import load_settings

settings = load_settings()
print(settings.db_host)

Returns: Settings instance

Raises: ValueError if required environment variables are missing

Source: src/config.py:36-49

MQTT Client

GatewayMqttClient

Main client that connects to MQTT broker, subscribes to flow topics, and processes messages.

Source: src/mqtt_client.py:17-125

Constructor

GatewayMqttClient(settings: Settings, session_factory, logger: logging.Logger)

settings · Settings · required

Configuration instance from load_settings()

session_factory · required

SQLAlchemy session factory (e.g., from sessionmaker())

logger · logging.Logger · required

Logger instance for error and status messages

load_mqtt_server()

Returns the first enabled MQTT server from the database, prioritizing default servers.

server = client.load_mqtt_server()

Returns: MqttServer instance

Raises: RuntimeError if no enabled servers exist

Query: SELECT * FROM mqtt_servers WHERE enabled = 1 ORDER BY is_default DESC, id ASC LIMIT 1

Source: src/mqtt_client.py:25-35

load_flows()

Returns all enabled flows from the database.

flows = client.load_flows()

Returns: list[Flow]

Query: SELECT * FROM flows WHERE enabled = 1

Source: src/mqtt_client.py:37-40

reload_flows()

Reloads flows from the database and updates MQTT subscriptions dynamically.

client.reload_flows()

This method:

1. Fetches current enabled flows
2. Compares with previous subscriptions
3. Subscribes to new topics
4. Unsubscribes from removed topics

Source: src/mqtt_client.py:55-63

run_forever()

Main loop that connects to MQTT broker and processes messages indefinitely.

client.run_forever()

This method:

1. Loads MQTT server configuration
2. Loads initial flows
3. Connects to broker with authentication if configured
4. Starts MQTT loop
5. Periodically reloads flows every flows_reload_interval_seconds

Blocks until interrupted

Source: src/mqtt_client.py:103-125

on_connect()

Callback triggered when MQTT connection is established. Subscribes to all flow topics.

Source: src/mqtt_client.py:65-72

on_message()

Callback triggered when an MQTT message arrives. Validates JSON payload and routes to matching flows.

Behavior:

• Decodes payload as UTF-8 JSON object
• Matches message topic against flow patterns (supports wildcards)
• Calls process_flow_message() for each matching flow
• Logs errors without crashing

Source: src/mqtt_client.py:74-101

Message Processing

ProcessResult

Dataclass returned by process_flow_message() with processing metadata.

Source: src/processor.py:16-21

flow_code · string

The flow’s unique code identifier

msg_id · integer

The incremented message ID after processing

action · string

The action that was performed (STORE_DB or POST_ENDPOINT)

process_flow_message()

Processes a single MQTT message according to flow configuration.

from src.processor import process_flow_message

result = process_flow_message(
    session_factory=session_factory,
    settings=settings,
    logger=logger,
    flow_id=1,
    payload={"temperature": 23.5, "humidity": 65.2}
)

session_factory · required

SQLAlchemy session factory

settings · Settings · required

Configuration instance

logger · logging.Logger · required

Logger for error messages

flow_id · integer · required

Database ID of the flow to process

payload · dict[str, Any] · required

Parsed JSON payload from MQTT message

Returns: ProcessResult

Raises: ValueError if payload validation fails or endpoint_url is missing for POST_ENDPOINT

Processing steps:

1. Locks the flow row with SELECT ... FOR UPDATE
2. Validates payload against payload_schema using validate_payload()
3. Increments last_msg_id
4. For STORE_DB: Creates DataRecord entries for each payload field
5. For POST_ENDPOINT: POSTs JSON payload to endpoint_url

Source: src/processor.py:79-135

validate_payload()

Validates that payload matches the expected schema.

from src.processor import validate_payload

validate_payload(
    payload={"temperature": 23.5, "humidity": 65.2},
    payload_schema={"temperature": "float", "humidity": "float"}
)

payload · dict[str, Any] · required

The message payload to validate

payload_schema · dict[str, Any] · required

Expected structure with field names and types

Raises: ValueError with descriptive message if:

• Required field is missing
• Field has incorrect type

Supported type mappings:

• string, str → Python str
• number, float → Python int or float
• int, integer → Python int
• bool, boolean → Python bool
• object → Python dict
• array → Python list

Source: src/processor.py:56-71