Skip to content

Utility Functions

cymongoose provides utility functions for JSON parsing, URL parsing, pattern matching, hashing, encoding, and more.

JSON Utilities

Mongoose includes a lightweight JSON parser. These functions extract values from JSON without full parsing.

json_get

cymongoose.json_get(data, path)

Extract a value from JSON by path (e.g., '$.user.name').

Parameters:

Name Type Description Default
data Union[str, bytes]

JSON string or bytes

required
path str

JSON path (e.g., '$.items[0].id')

required

Returns:

Type Description
Optional[str]

String value at path, or None if not found

Source code in src/cymongoose/_mongoose.pyi
def json_get(data: Union[str, bytes], path: str) -> Optional[str]:
    """Extract a value from JSON by path (e.g., '$.user.name').

    Args:
        data: JSON string or bytes
        path: JSON path (e.g., '$.items[0].id')

    Returns:
        String value at path, or None if not found
    """
    ...

Example:

json_str = '{"user": {"name": "Alice", "age": 30}, "items": [1, 2, 3]}'

# Get nested value
name = json_get(json_str, "$.user.name")  # "Alice"

# Get array element
first = json_get(json_str, "$.items[0]")  # "1"

json_get_num

cymongoose.json_get_num(data, path, default=None)

Extract a numeric value from JSON by path.

Parameters:

Name Type Description Default
data Union[str, bytes]

JSON string or bytes

required
path str

JSON path (e.g., '$.count')

required
default Optional[float]

Default value if not found or not a number

None

Returns:

Type Description
Optional[float]

Float value at path, or default if not found

Source code in src/cymongoose/_mongoose.pyi
def json_get_num(
    data: Union[str, bytes], path: str, default: Optional[float] = None
) -> Optional[float]:
    """Extract a numeric value from JSON by path.

    Args:
        data: JSON string or bytes
        path: JSON path (e.g., '$.count')
        default: Default value if not found or not a number

    Returns:
        Float value at path, or default if not found
    """
    ...

Example:

json_str = '{"temperature": 23.5, "humidity": 65}'

temp = json_get_num(json_str, "$.temperature")  # 23.5
pressure = json_get_num(json_str, "$.pressure", default=0.0)  # 0.0

json_get_bool

cymongoose.json_get_bool(data, path, default=None)

Extract a boolean value from JSON by path.

Parameters:

Name Type Description Default
data Union[str, bytes]

JSON string or bytes

required
path str

JSON path (e.g., '$.enabled')

required
default Optional[bool]

Default value if not found or not a boolean

None

Returns:

Type Description
Optional[bool]

Boolean value at path, or default if not found

Source code in src/cymongoose/_mongoose.pyi
def json_get_bool(
    data: Union[str, bytes], path: str, default: Optional[bool] = None
) -> Optional[bool]:
    """Extract a boolean value from JSON by path.

    Args:
        data: JSON string or bytes
        path: JSON path (e.g., '$.enabled')
        default: Default value if not found or not a boolean

    Returns:
        Boolean value at path, or default if not found
    """
    ...

Example:

json_str = '{"enabled": true, "debug": false}'

enabled = json_get_bool(json_str, "$.enabled")  # True
debug = json_get_bool(json_str, "$.debug")  # False
missing = json_get_bool(json_str, "$.other", default=False)  # False

json_get_long

cymongoose.json_get_long(data, path, default=0)

Extract an integer value from JSON by path.

Parameters:

Name Type Description Default
data Union[str, bytes]

JSON string or bytes

required
path str

JSON path (e.g., '$.id')

required
default int

Default value if not found or not an integer

0

Returns:

Type Description
int

Integer value at path, or default if not found

Source code in src/cymongoose/_mongoose.pyi
def json_get_long(data: Union[str, bytes], path: str, default: int = 0) -> int:
    """Extract an integer value from JSON by path.

    Args:
        data: JSON string or bytes
        path: JSON path (e.g., '$.id')
        default: Default value if not found or not an integer

    Returns:
        Integer value at path, or default if not found
    """
    ...

Example:

json_str = '{"count": 12345, "id": 9876543210}'

count = json_get_long(json_str, "$.count")  # 12345
user_id = json_get_long(json_str, "$.id")  # 9876543210
missing = json_get_long(json_str, "$.other", default=0)  # 0

json_get_str

cymongoose.json_get_str(data, path)

Extract a string value from JSON by path (automatically unescapes).

Parameters:

Name Type Description Default
data Union[str, bytes]

JSON string or bytes

required
path str

JSON path (e.g., '$.message')

required

Returns:

Type Description
Optional[str]

Unescaped string value at path, or None if not found

Source code in src/cymongoose/_mongoose.pyi
def json_get_str(data: Union[str, bytes], path: str) -> Optional[str]:
    """Extract a string value from JSON by path (automatically unescapes).

    Args:
        data: JSON string or bytes
        path: JSON path (e.g., '$.message')

    Returns:
        Unescaped string value at path, or None if not found
    """
    ...

Example:

json_str = '{"message": "Hello, World!", "path": "/home/user"}'

# Automatically unescapes JSON strings
message = json_get_str(json_str, "$.message")  # "Hello, World!"
path = json_get_str(json_str, "$.path")  # "/home/user"

Complete JSON Parsing Example

from cymongoose import (
    json_get,
    json_get_num,
    json_get_bool,
    json_get_long,
    json_get_str,
)

def handler(conn, ev, data):
    if ev == MG_EV_HTTP_MSG:
        json_body = data.body_text

        # Parse different types
        user_id = json_get_long(json_body, "$.user.id")
        username = json_get_str(json_body, "$.user.name")
        age = json_get_num(json_body, "$.user.age")
        active = json_get_bool(json_body, "$.user.active")

        # Build response
        response = {
            "id": user_id,
            "name": username,
            "age": age,
            "active": active,
        }

        import json
        conn.reply(200, json.dumps(response).encode())

URL Encoding and Parsing

url_encode

cymongoose.url_encode(data)

URL-encode a string.

Parameters:

Name Type Description Default
data str

String to encode

required

Returns:

Type Description
str

URL-encoded string

Source code in src/cymongoose/_mongoose.pyi
def url_encode(data: str) -> str:
    """URL-encode a string.

    Args:
        data: String to encode

    Returns:
        URL-encoded string
    """
    ...

Example:

# Encode query parameters
param = url_encode("hello world")  # "hello%20world"
email = url_encode("user@example.com")  # "user%40example.com"

# Build query string
query = f"name={url_encode(name)}&email={url_encode(email)}"

# Make request
url = f"http://example.com/api?{query}"
conn = manager.connect(url, http=True)

url_port

cymongoose.url_port(url)

Return the port number from a URL, or default port for the scheme.

Source code in src/cymongoose/_mongoose.pyi
def url_port(url: str) -> int:
    """Return the port number from a URL, or default port for the scheme."""
    ...

url_host

cymongoose.url_host(url)

Return the host component of a URL.

Source code in src/cymongoose/_mongoose.pyi
def url_host(url: str) -> str:
    """Return the host component of a URL."""
    ...

url_user

cymongoose.url_user(url)

Return the username from a URL, or empty string if absent.

Source code in src/cymongoose/_mongoose.pyi
def url_user(url: str) -> str:
    """Return the username from a URL, or empty string if absent."""
    ...

url_pass

cymongoose.url_pass(url)

Return the password from a URL, or empty string if absent.

Source code in src/cymongoose/_mongoose.pyi
def url_pass(url: str) -> str:
    """Return the password from a URL, or empty string if absent."""
    ...

url_uri

cymongoose.url_uri(url)

Return the URI (path + query) from a URL.

Source code in src/cymongoose/_mongoose.pyi
def url_uri(url: str) -> str:
    """Return the URI (path + query) from a URL."""
    ...

url_is_ssl

cymongoose.url_is_ssl(url)

Return True if the URL scheme implies TLS.

Source code in src/cymongoose/_mongoose.pyi
def url_is_ssl(url: str) -> bool:
    """Return True if the URL scheme implies TLS."""
    ...

Example:

from cymongoose import url_port, url_host, url_user, url_pass, url_uri, url_is_ssl

url = "https://admin:secret@example.com:8443/api/v1?key=abc"

url_host(url)    # "example.com"
url_port(url)    # 8443
url_user(url)    # "admin"
url_pass(url)    # "secret"
url_uri(url)     # "/api/v1?key=abc"
url_is_ssl(url)  # True

# Default ports
url_port("http://example.com")   # 80
url_port("https://example.com")  # 443
url_port("mqtt://broker.com")    # 1883

HTTP Variable Extraction

http_var

cymongoose.http_var(buf, name)

Extract a form/query variable from a buffer.

Parameters:

Name Type Description Default
buf Union[str, bytes]

Query string or form body

required
name str

Variable name

required

Returns:

Type Description
Optional[str]

Value string, or None if not found

Source code in src/cymongoose/_mongoose.pyi
def http_var(buf: Union[str, bytes], name: str) -> Optional[str]:
    """Extract a form/query variable from a buffer.

    Args:
        buf: Query string or form body
        name: Variable name

    Returns:
        Value string, or None if not found
    """
    ...

Example:

from cymongoose import http_var

# Extract from query string
http_var("name=Alice&age=30", "name")  # "Alice"
http_var("name=Alice&age=30", "age")   # "30"
http_var("name=Alice", "missing")      # None

# In an HTTP handler, pass the query or body directly
def handler(conn, ev, data):
    if ev == MG_EV_HTTP_MSG:
        user = http_var(data.query, "user")

Pattern Matching

match

cymongoose.match(s, pattern)

Match a string against a mongoose glob pattern.

Pattern syntax: ? (single char), * (segment), # (greedy across /).

Parameters:

Name Type Description Default
s str

String to match

required
pattern str

Glob pattern

required

Returns:

Type Description
Tuple[bool, List[str]]

(matched, captures) tuple

Source code in src/cymongoose/_mongoose.pyi
def match(s: str, pattern: str) -> Tuple[bool, List[str]]:
    """Match a string against a mongoose glob pattern.

    Pattern syntax: ? (single char), * (segment), # (greedy across /).

    Args:
        s: String to match
        pattern: Glob pattern

    Returns:
        (matched, captures) tuple
    """
    ...

Pattern syntax:

  • ? matches any single character (captures it)
  • * matches zero or more characters except / (captures them)
  • # matches zero or more characters including / (captures them)

Example:

from cymongoose import match

# Exact match
match("hello", "hello")          # (True, [])

# Wildcard captures
match("/api/users", "/api/*")    # (True, ["users"])
match("/a/b/c", "#")             # (True, ["/a/b/c"])

# Route matching with multiple captures
match("/users/42", "/users/??"  )  # (True, ["4", "2"])
match("/api/v1/items", "/api/*/items")  # (True, ["v1"])

# No match
match("foo/bar", "*")            # (False, []) -- * doesn't cross /

JSON-RPC Framework

Rpc

cymongoose.Rpc

JSON-RPC dispatcher backed by mongoose's built-in RPC framework.

Source code in src/cymongoose/_mongoose.pyi
class Rpc:
    """JSON-RPC dispatcher backed by mongoose's built-in RPC framework."""

    def add(self, method: str, handler: Callable[[RpcReq], None]) -> None:
        """Register an RPC method handler.

        Args:
            method: Method name or pattern
            handler: Callable receiving an RpcReq argument
        """
        ...

    def process(self, frame: Union[str, bytes]) -> str:
        """Dispatch a JSON-RPC request frame and return the response.

        Args:
            frame: JSON-RPC request string (or bytes)

        Returns:
            JSON-RPC response string, or empty string for notifications
        """
        ...

    @property
    def methods(self) -> List[str]:
        """Return list of registered method patterns."""
        ...

methods property

Return list of registered method patterns.

add(method, handler)

Register an RPC method handler.

Parameters:

Name Type Description Default
method str

Method name or pattern

required
handler Callable[[RpcReq], None]

Callable receiving an RpcReq argument

required
Source code in src/cymongoose/_mongoose.pyi
def add(self, method: str, handler: Callable[[RpcReq], None]) -> None:
    """Register an RPC method handler.

    Args:
        method: Method name or pattern
        handler: Callable receiving an RpcReq argument
    """
    ...

process(frame)

Dispatch a JSON-RPC request frame and return the response.

Parameters:

Name Type Description Default
frame Union[str, bytes]

JSON-RPC request string (or bytes)

required

Returns:

Type Description
str

JSON-RPC response string, or empty string for notifications

Source code in src/cymongoose/_mongoose.pyi
def process(self, frame: Union[str, bytes]) -> str:
    """Dispatch a JSON-RPC request frame and return the response.

    Args:
        frame: JSON-RPC request string (or bytes)

    Returns:
        JSON-RPC response string, or empty string for notifications
    """
    ...

RpcReq

cymongoose.RpcReq

Request object passed to RPC handler callbacks.

Source code in src/cymongoose/_mongoose.pyi
class RpcReq:
    """Request object passed to RPC handler callbacks."""

    @property
    def frame(self) -> str:
        """The raw JSON-RPC request frame."""
        ...

    def ok(self, result_json: str) -> None:
        """Send a successful JSON-RPC response.

        Args:
            result_json: Raw JSON value for the "result" field.
        """
        ...

    def err(self, code: int, message: str) -> None:
        """Send a JSON-RPC error response.

        Args:
            code: JSON-RPC error code
            message: Human-readable error message
        """
        ...

frame property

The raw JSON-RPC request frame.

ok(result_json)

Send a successful JSON-RPC response.

Parameters:

Name Type Description Default
result_json str

Raw JSON value for the "result" field.

required
Source code in src/cymongoose/_mongoose.pyi
def ok(self, result_json: str) -> None:
    """Send a successful JSON-RPC response.

    Args:
        result_json: Raw JSON value for the "result" field.
    """
    ...

err(code, message)

Send a JSON-RPC error response.

Parameters:

Name Type Description Default
code int

JSON-RPC error code

required
message str

Human-readable error message

required
Source code in src/cymongoose/_mongoose.pyi
def err(self, code: int, message: str) -> None:
    """Send a JSON-RPC error response.

    Args:
        code: JSON-RPC error code
        message: Human-readable error message
    """
    ...

Example:

import json
from cymongoose import Rpc, RpcReq, json_get_num, json_get_str, MG_EV_HTTP_MSG

rpc = Rpc()

def add(req):
    a = json_get_num(req.frame, "$.params[0]")
    b = json_get_num(req.frame, "$.params[1]")
    req.ok(str(a + b))

def greet(req):
    name = json_get_str(req.frame, "$.params[0]")
    req.ok(json.dumps(f"Hello, {name}!"))

def fail(req):
    req.err(-32000, "something went wrong")

rpc.add("add", add)
rpc.add("greet", greet)
rpc.add("fail", fail)

print(rpc.methods)  # ["fail", "greet", "add"]

def handler(conn, ev, data):
    if ev == MG_EV_HTTP_MSG and data.uri == "/rpc":
        response = rpc.process(data.body_text)
        conn.reply(200, response, {"Content-Type": "application/json"})
        conn.drain()

Hashing

md5

cymongoose.md5(data)

Compute MD5 hash. Returns 16-byte digest.

Source code in src/cymongoose/_mongoose.pyi
def md5(data: Union[str, bytes]) -> bytes:
    """Compute MD5 hash. Returns 16-byte digest."""
    ...

sha1

cymongoose.sha1(data)

Compute SHA1 hash. Returns 20-byte digest.

Source code in src/cymongoose/_mongoose.pyi
def sha1(data: Union[str, bytes]) -> bytes:
    """Compute SHA1 hash. Returns 20-byte digest."""
    ...

sha256

cymongoose.sha256(data)

Compute SHA256 hash. Returns 32-byte digest.

Source code in src/cymongoose/_mongoose.pyi
def sha256(data: Union[str, bytes]) -> bytes:
    """Compute SHA256 hash. Returns 32-byte digest."""
    ...

hmac_sha256

cymongoose.hmac_sha256(key, data)

Compute HMAC-SHA256. Returns 32-byte digest.

Source code in src/cymongoose/_mongoose.pyi
def hmac_sha256(key: Union[str, bytes], data: Union[str, bytes]) -> bytes:
    """Compute HMAC-SHA256. Returns 32-byte digest."""
    ...

Example:

from cymongoose import md5, sha1, sha256, hmac_sha256

# Hash data (accepts str or bytes)
md5("hello")           # 16-byte digest
sha1("hello")          # 20-byte digest
sha256("hello")        # 32-byte digest

# HMAC for message authentication
sig = hmac_sha256("secret-key", "message-to-sign")

# Hex representation
sha256(b"hello").hex()  # "2cf24dba5fb0a30e..."

Base64

base64_encode

cymongoose.base64_encode(data)

Base64-encode data. Returns encoded string.

Source code in src/cymongoose/_mongoose.pyi
def base64_encode(data: Union[str, bytes]) -> str:
    """Base64-encode data. Returns encoded string."""
    ...

base64_decode

cymongoose.base64_decode(data)

Base64-decode a string. Returns decoded bytes.

Source code in src/cymongoose/_mongoose.pyi
def base64_decode(data: str) -> bytes:
    """Base64-decode a string. Returns decoded bytes."""
    ...

Example:

from cymongoose import base64_encode, base64_decode

encoded = base64_encode(b"Hello, World!")  # "SGVsbG8sIFdvcmxkIQ=="
decoded = base64_decode(encoded)            # b"Hello, World!"

# Works with str input too
base64_encode("binary data")

Misc Utilities

millis

cymongoose.millis()

Return milliseconds since boot (monotonic clock).

Source code in src/cymongoose/_mongoose.pyi
def millis() -> int:
    """Return milliseconds since boot (monotonic clock)."""
    ...

random_bytes

cymongoose.random_bytes(length)

Generate cryptographically random bytes.

Source code in src/cymongoose/_mongoose.pyi
def random_bytes(length: int) -> bytes:
    """Generate cryptographically random bytes."""
    ...

random_str

cymongoose.random_str(length)

Generate a random alphanumeric string.

Source code in src/cymongoose/_mongoose.pyi
def random_str(length: int) -> str:
    """Generate a random alphanumeric string."""
    ...

crc32

cymongoose.crc32(data, initial=0)

Compute CRC32 checksum.

Source code in src/cymongoose/_mongoose.pyi
def crc32(data: Union[str, bytes], initial: int = 0) -> int:
    """Compute CRC32 checksum."""
    ...

Example:

from cymongoose import millis, random_bytes, random_str, crc32

# Monotonic time
start = millis()
# ... do work ...
elapsed = millis() - start

# Random data
token = random_str(32)       # e.g. "aB3kM9xQ..."
nonce = random_bytes(16)     # 16 random bytes

# CRC32 checksum
checksum = crc32(b"hello")
# Incremental
crc = crc32(b"hel")
crc = crc32(b"lo", crc)     # same as crc32(b"hello")

Multipart Form Data

http_parse_multipart

cymongoose.http_parse_multipart(body, offset=0)

Parse the next multipart form part from HTTP body.

Parameters:

Name Type Description Default
body Union[str, bytes]

HTTP body (str or bytes)

required
offset int

Offset to start parsing from

0

Returns:

Type Description
int

Tuple of (next_offset, part_dict) where part_dict contains:

Optional[Dict[str, Union[str, bytes]]]
  • 'name': form field name
Tuple[int, Optional[Dict[str, Union[str, bytes]]]]
  • 'filename': filename (if file upload)
Tuple[int, Optional[Dict[str, Union[str, bytes]]]]
  • 'body': part data as bytes
Tuple[int, Optional[Dict[str, Union[str, bytes]]]]

Returns (0, None) if no more parts.

Source code in src/cymongoose/_mongoose.pyi
def http_parse_multipart(
    body: Union[str, bytes], offset: int = 0
) -> Tuple[int, Optional[Dict[str, Union[str, bytes]]]]:
    """Parse the next multipart form part from HTTP body.

    Args:
        body: HTTP body (str or bytes)
        offset: Offset to start parsing from

    Returns:
        Tuple of (next_offset, part_dict) where part_dict contains:
        - 'name': form field name
        - 'filename': filename (if file upload)
        - 'body': part data as bytes
        Returns (0, None) if no more parts.
    """
    ...

Example:

from cymongoose import MG_EV_HTTP_MSG, http_parse_multipart

def handler(conn, ev, data):
    if ev == MG_EV_HTTP_MSG and data.method == "POST":
        # Parse multipart form data
        offset = 0
        while True:
            offset, part = http_parse_multipart(data.body_bytes, offset)
            if part is None:
                break  # No more parts

            # Field name
            field_name = part['name']

            # File upload
            if part['filename']:
                filename = part['filename']
                file_data = part['body']
                print(f"File upload: {filename} ({len(file_data)} bytes)")

                # Save file
                with open(f"uploads/{filename}", "wb") as f:
                    f.write(file_data)
            else:
                # Regular form field
                field_value = part['body'].decode('utf-8')
                print(f"Field: {field_name} = {field_value}")

        conn.reply(200, b'{"status": "uploaded"}')
        conn.drain()

File Upload Server Example

import os
from cymongoose import Manager, MG_EV_HTTP_MSG, http_parse_multipart

def handler(conn, ev, data):
    if ev == MG_EV_HTTP_MSG:
        if data.method == "POST" and data.uri == "/upload":
            os.makedirs("uploads", exist_ok=True)

            offset = 0
            uploaded_files = []

            while True:
                offset, part = http_parse_multipart(data.body_bytes, offset)
                if part is None:
                    break

                if part['filename']:
                    # Save uploaded file
                    filename = part['filename']
                    filepath = os.path.join("uploads", filename)

                    with open(filepath, "wb") as f:
                        f.write(part['body'])

                    uploaded_files.append(filename)
                    print(f"Saved: {filepath}")

            # Return success response
            import json
            response = {
                "status": "success",
                "files": uploaded_files,
            }
            conn.reply(200, json.dumps(response).encode(),
                      headers={"Content-Type": "application/json"})
        else:
            # Serve upload form
            html = b"""
            <html>
            <body>
                <h1>File Upload</h1>
                <form method="POST" action="/upload"
                      enctype="multipart/form-data">
                    <input type="file" name="files" multiple>
                    <button type="submit">Upload</button>
                </form>
            </body>
            </html>
            """
            conn.reply(200, html,
                      headers={"Content-Type": "text/html"})
        conn.drain()

Event Debugging

event_name

cymongoose.event_name(ev)

Return a human-readable name for a Mongoose event constant.

Parameters:

Name Type Description Default
ev int

Event constant (e.g., MG_EV_HTTP_MSG)

required

Returns:

Type Description
str

Event name string (e.g., "MG_EV_HTTP_MSG"). Returns

str

"MG_EV_USER+N" for user events and "MG_EV_UNKNOWN(N)" for

str

unrecognised values.

Source code in src/cymongoose/_mongoose.pyi
def event_name(ev: int) -> str:
    """Return a human-readable name for a Mongoose event constant.

    Args:
        ev: Event constant (e.g., MG_EV_HTTP_MSG)

    Returns:
        Event name string (e.g., "MG_EV_HTTP_MSG"). Returns
        "MG_EV_USER+N" for user events and "MG_EV_UNKNOWN(N)" for
        unrecognised values.
    """
    ...

Example:

from cymongoose import event_name, MG_EV_HTTP_MSG

def handler(conn, ev, data):
    print(f"Event: {event_name(ev)}")  # "MG_EV_HTTP_MSG"

Logging Control

Control the Mongoose C library's internal logging.

log_set

cymongoose.log_set(level)

Set the Mongoose C library log verbosity level.

Parameters:

Name Type Description Default
level int

One of MG_LL_NONE (0), MG_LL_ERROR (1), MG_LL_INFO (2), MG_LL_DEBUG (3), MG_LL_VERBOSE (4)

required
Source code in src/cymongoose/_mongoose.pyi
def log_set(level: int) -> None:
    """Set the Mongoose C library log verbosity level.

    Args:
        level: One of MG_LL_NONE (0), MG_LL_ERROR (1), MG_LL_INFO (2),
               MG_LL_DEBUG (3), MG_LL_VERBOSE (4)
    """
    ...

log_get

cymongoose.log_get()

Return the current Mongoose C library log verbosity level.

Source code in src/cymongoose/_mongoose.pyi
def log_get() -> int:
    """Return the current Mongoose C library log verbosity level."""
    ...

Example:

from cymongoose import log_set, log_get, MG_LL_DEBUG, MG_LL_NONE

# Enable debug logging
log_set(MG_LL_DEBUG)

# Check current level
print(log_get())  # 3

# Disable logging
log_set(MG_LL_NONE)

Constants

Event Types

See Guide for event handling details.

from cymongoose import (
    MG_EV_ERROR,
    MG_EV_OPEN,
    MG_EV_POLL,
    MG_EV_RESOLVE,
    MG_EV_CONNECT,
    MG_EV_ACCEPT,
    MG_EV_TLS_HS,
    MG_EV_READ,
    MG_EV_WRITE,
    MG_EV_CLOSE,
    MG_EV_HTTP_HDRS,
    MG_EV_HTTP_MSG,
    MG_EV_WS_OPEN,
    MG_EV_WS_MSG,
    MG_EV_WS_CTL,
    MG_EV_MQTT_CMD,
    MG_EV_MQTT_MSG,
    MG_EV_MQTT_OPEN,
    MG_EV_SNTP_TIME,
    MG_EV_WAKEUP,
    MG_EV_USER,
)

MQTT v5 Property Types

from cymongoose import (
    MQTT_PROP_TYPE_BYTE,
    MQTT_PROP_TYPE_SHORT,
    MQTT_PROP_TYPE_INT,
    MQTT_PROP_TYPE_VARIABLE_INT,
    MQTT_PROP_TYPE_STRING,
    MQTT_PROP_TYPE_STRING_PAIR,
    MQTT_PROP_TYPE_BINARY_DATA,
)

WebSocket Opcodes

from cymongoose import (
    WEBSOCKET_OP_TEXT,
    WEBSOCKET_OP_BINARY,
    WEBSOCKET_OP_PING,
    WEBSOCKET_OP_PONG,
)

# Use with ws_send()
conn.ws_send("Hello", WEBSOCKET_OP_TEXT)
conn.ws_send(b"\x00\x01\x02", WEBSOCKET_OP_BINARY)

See Also

  • HttpMessage - HTTP message access
  • Connection - Connection methods
  • Examples - Complete examples