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
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
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
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
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
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 |
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)
¶
url_host¶
cymongoose.url_host(url)
¶
url_user¶
cymongoose.url_user(url)
¶
url_pass¶
cymongoose.url_pass(url)
¶
url_uri¶
cymongoose.url_uri(url)
¶
url_is_ssl¶
cymongoose.url_is_ssl(url)
¶
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
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
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
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 |
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
RpcReq¶
cymongoose.RpcReq
¶
Request object passed to RPC handler callbacks.
Source code in src/cymongoose/_mongoose.pyi
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 |
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 |
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)
¶
sha1¶
cymongoose.sha1(data)
¶
sha256¶
cymongoose.sha256(data)
¶
hmac_sha256¶
cymongoose.hmac_sha256(key, data)
¶
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_decode¶
cymongoose.base64_decode(data)
¶
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()
¶
random_bytes¶
cymongoose.random_bytes(length)
¶
random_str¶
cymongoose.random_str(length)
¶
crc32¶
cymongoose.crc32(data, initial=0)
¶
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]]]
|
|
Tuple[int, Optional[Dict[str, Union[str, bytes]]]]
|
|
Tuple[int, Optional[Dict[str, Union[str, bytes]]]]
|
|
Tuple[int, Optional[Dict[str, Union[str, bytes]]]]
|
Returns (0, None) if no more parts. |
Source code in src/cymongoose/_mongoose.pyi
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
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 |
log_get¶
cymongoose.log_get()
¶
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 accessConnection- Connection methods- Examples - Complete examples