Utility Functions

cymongoose provides utility functions for JSON parsing, URL encoding, and multipart form handling.

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

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)

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,
)

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