genro-toolbox 0.8.0

Essential utilities for Genro Kyō

Genro-Toolbox

Essential utilities for the Genro ecosystem

Part of Genro Kyō ecosystem.

A lightweight, zero-dependency Python library providing core utilities that can be used across all Genro projects.

📚 Full Documentation

Installation

pip install genro-toolbox

Features

• SmartOptions - Multi-source config with merge via + operator
• TreeDict - Hierarchical dict with dot notation and path access
• DictObj - Dict subclass with dot-access for attribute-style read/write
• extract_kwargs - Decorator to group kwargs by prefix
• dictExtract - Extract dict items by key prefix
• smartsplit - Split strings honoring escaped separators
• get_uuid - Sortable 22-char unique identifiers for distributed systems
• smartasync - Unified sync/async API with automatic context detection
• smarttimer - Non-blocking timers (setTimeout/setInterval) with sync/async detection
• safe_is_instance - isinstance() without importing the class
• render_ascii_table - ASCII table rendering with formatting
• render_markdown_table - Markdown table rendering
• tags_match - Boolean expression matcher for tag-based filtering

Examples

SmartOptions

Load config from multiple sources with type conversion:

from genro_toolbox import SmartOptions
import sys

def serve(host: str = '127.0.0.1', port: int = 8000, debug: bool = False):
    pass

# Load from env and argv with automatic type conversion
# Priority: defaults < env < argv
config = SmartOptions(serve, env='MYAPP', argv=sys.argv[1:])

config["host"]   # from env (MYAPP_HOST) or default
config["port"]   # int from env (MYAPP_PORT) or argv
config["debug"]  # True if --debug passed

Compose with + for file configs:

config = (
    SmartOptions('config.yaml') +      # file config
    SmartOptions(serve, env='MYAPP', argv=sys.argv[1:])  # defaults < env < argv
)

Load from files (YAML, JSON, TOML, INI):

opts = SmartOptions('config.yaml')
opts["server.host"]  # nested dicts become SmartOptions
opts["middleware.cors"]  # string lists become feature flags (True)
opts["apps.shop.module"]  # list of dicts indexed by first key

Basic merge with filtering:

opts = SmartOptions(
    {"timeout": 30},                    # runtime values
    {"timeout": 10, "retries": 3},      # defaults
    ignore_none=True,
    ignore_empty=True,
)

opts["timeout"]   # 30 (runtime wins)
opts["retries"]   # 3 (from defaults)

TreeDict

Hierarchical dictionary with path access:

from genro_toolbox import TreeDict

# Create from nested dict
td = TreeDict({"user": {"name": "Alice", "prefs": {"theme": "dark"}}})

# Or from JSON string
td = TreeDict('{"user": {"name": "Alice"}}')

# Or from config file (JSON, YAML, TOML, INI)
td = TreeDict.from_file("config.yaml")

# Path string access
td["user.name"]        # "Alice"
td["user.prefs.theme"] # "dark"
td["missing"]          # None (no KeyError)

# Auto-create intermediate dicts on write
td["settings.db.host"] = "localhost"
td["settings.db.host"] # "localhost"

# List access with #N syntax
td = TreeDict({"users": [{"name": "Alice"}, {"name": "Bob"}]})
td["users.#0.name"]    # "Alice"
td["users.#1.name"]    # "Bob"

# Walk all paths
for path, value in td.walk():
    print(f"{path} = {value}")
# users.#0.name = Alice
# users.#1.name = Bob

# Thread-safe access (sync)
with td:
    td["counter"] = td["counter"] + 1

# Async-safe access
async with td:
    td["counter"] = td["counter"] + 1

# Convert back to dict
td.as_dict()  # {"user": {"name": "Alice", ...}}

extract_kwargs Decorator

from genro_toolbox import extract_kwargs

@extract_kwargs(logging=True, cache=True)
def my_function(name, logging_kwargs=None, cache_kwargs=None, **kwargs):
    print(f"logging: {logging_kwargs}")
    print(f"cache: {cache_kwargs}")

my_function(
    "test",
    logging_level="INFO",
    logging_format="json",
    cache_ttl=300,
)
# logging: {'level': 'INFO', 'format': 'json'}
# cache: {'ttl': 300}

safe_is_instance

from genro_toolbox import safe_is_instance

# Check type without importing
safe_is_instance(42, "builtins.int")              # True
safe_is_instance(my_obj, "mypackage.BaseClass")   # True (includes subclasses)

ASCII & Markdown Tables

from genro_toolbox import render_ascii_table, render_markdown_table

data = {
    "headers": [
        {"name": "Name", "type": "str"},
        {"name": "Active", "type": "bool"},
    ],
    "rows": [
        ["Alice", "yes"],
        ["Bob", "no"],
    ],
}

print(render_ascii_table(data))
# +-------+--------+
# |Name   |Active  |
# +-------+--------+
# |Alice  |true    |
# +-------+--------+
# |Bob    |false   |
# +-------+--------+

tags_match

from genro_toolbox import tags_match

# Simple tag check
tags_match("admin", {"admin", "user"})  # True

# OR (comma, pipe, or keyword)
tags_match("admin,public", {"public"})  # True
tags_match("admin or public", {"admin"})  # True

# AND (ampersand or keyword)
tags_match("admin&internal", {"admin", "internal"})  # True
tags_match("admin and internal", {"admin"})  # False

# NOT (exclamation or keyword)
tags_match("!admin", {"public"})  # True
tags_match("not admin", {"admin"})  # False

# Complex expressions
tags_match("(admin|public)&!internal", {"admin"})  # True
tags_match("(admin or public) and not internal", {"admin", "internal"})  # False

get_uuid

from genro_toolbox import get_uuid

# Generate sortable unique identifiers
uid = get_uuid()  # e.g., "Z00005KmLxHj7F9aGbCd3e"

len(uid)      # 22 characters
uid[0]        # 'Z' (version marker, sorts after legacy UUIDs)
uid.isalnum() # True (URL-safe)

# IDs are lexicographically sortable by creation time
ids = [get_uuid() for _ in range(3)]
sorted(ids) == ids  # True (already sorted)

smartasync

from genro_toolbox import smartasync

class DataManager:
    @smartasync
    async def fetch_data(self, url: str):
        async with httpx.AsyncClient() as client:
            return await client.get(url).json()

manager = DataManager()

# Sync context - no await needed!
data = manager.fetch_data("https://api.example.com")

# Async context - use await
async def main():
    data = await manager.fetch_data("https://api.example.com")

# Also works with sync methods in async context (offloaded to thread)
class LegacyProcessor:
    @smartasync
    def cpu_intensive(self, data):
        return process(data)  # Blocking operation

async def main():
    proc = LegacyProcessor()
    result = await proc.cpu_intensive(data)  # Won't block event loop

DictObj

from genro_toolbox import DictObj

ctx = DictObj()
ctx.db = connection
ctx.session = session_obj
ctx.db.execute(...)   # dot-access
"db" in ctx           # True (dict-access)
del ctx.session       # works too

smartsplit

from genro_toolbox import smartsplit

smartsplit("a.b.c", ".")          # ['a', 'b', 'c']
smartsplit(r"a\.b.c", ".")        # ['a\\.b', 'c']  (escaped separator preserved)
smartsplit("one , two , three", ",")  # ['one', 'two', 'three']  (strips whitespace)

dictExtract

from genro_toolbox import dictExtract

kwargs = {"logging_level": "INFO", "logging_format": "json", "cache_ttl": 300}
dictExtract(kwargs, "logging_")  # {'level': 'INFO', 'format': 'json'}

smarttimer

Non-blocking timers with automatic sync/async detection:

from genro_toolbox import set_timeout, set_interval, cancel_timer

# Token refresh: renew 5 min before expiry (inside a server)
set_timeout(expires_in - 300, refresh_token)

# Heartbeat: ping every 30s (works in both sync and async)
hb = set_interval(30.0, ws.send_json, {"type": "ping"})

# Cancel when done
cancel_timer(hb)

# Async callbacks work transparently
async def check_job(job_id):
    status = await api.get_status(job_id)
    if status == "completed":
        cancel_timer(pollers[job_id])

pollers = {}
pollers["j1"] = set_interval(5.0, check_job, "j1")

Philosophy

If you write a generic helper that could be useful elsewhere, put it in genro-toolbox.

This library serves as the foundation for utilities shared across:

• genro-asgi
• genro-routes
• genro-api
• Other Genro Kyō projects

License

Apache License 2.0 - See LICENSE for details.

Copyright 2025 Softwell S.r.l.