muga 0.1.3b0

Observability for developers who live in the terminal.

muga (Python SDK)

Muga observability SDK for Python. Ships logs to Muga via OpenTelemetry OTLP, captures uncaught exceptions, instruments Flask/FastAPI, and pings cron monitors.

Install

pip install muga

Framework extras:

pip install 'muga[flask]'
pip install 'muga[fastapi]'

Set your project token:

export MUGA_TOKEN=muga_xxxxxxxxxxxx

Quickstart

from muga import init, MugaLogger

init(service_name="api")

log = MugaLogger("api")
log.info("server started", {"port": "3000"})

init() reads MUGA_TOKEN and MUGA_ENDPOINT from env (or accepts token= / endpoint=). When the token is missing, init() does not raise — it prints a single warning to stderr and returns. MugaLogger calls and the framework middlewares stay importable; their emits become no-ops via OpenTelemetry's default NoOpLoggerProvider. This keeps dev/CI runs unblocked when no token is configured.

On successful init the resolved endpoint and service name are printed to stderr so misconfigurations (wrong MUGA_ENDPOINT, fallback to default) are visible without enabling debug=True.

Auto-exception capture

init() installs sys.excepthook, threading.excepthook, and an asyncio loop exception handler by default. They emit a FATAL/ERROR log with exception.type, exception.message, and exception.stacktrace attributes, force-flush the batch processor so the record drains before the interpreter exits, then delegate to the runtime's default behaviour.

Behaviour matrix

Source	Severity	Triggered by
sys.excepthook	FATAL	Uncaught exception on the main thread.
threading.excepthook	ERROR	Uncaught exception in a worker thread.
asyncio loop handler	ERROR	Unhandled exception on a task in a loop attached at init() time.

Caveats:

• The asyncio handler is only attached to the loop that exists when init() runs. If your code creates a new loop later (asyncio.run(...), a fresh asyncio.new_event_loop()), call install_exception_handlers(loop=...) from inside that loop, or re-call init() once the loop is current.
• All three hooks are idempotent — calling init() twice does not double-install.

Opt out:

init(capture_exceptions=False)

Flask

from flask import Flask
from muga import init
from muga.flask import muga_flask

init()
app = Flask(__name__)
muga_flask(app)

One log per request: INFO for <500, ERROR for >=500. Attributes: http.method, http.path, http.status_code, http.duration_ms, http.user_agent.

FastAPI

from fastapi import FastAPI
from muga import init
from muga.fastapi import MugaFastAPIMiddleware

init()
app = FastAPI()
app.add_middleware(MugaFastAPIMiddleware)

Same attribute set as the Flask middleware.

Cron heartbeat

Stand-alone helper. Does not require init().

from muga import heartbeat

heartbeat("daily-cleanup")

POSTs to <endpoint>/v1/crons/daily-cleanup/ping with Authorization: Bearer $MUGA_TOKEN. Raises on missing token or non-2xx response. The cron name is URL-encoded.

For async code, wrap in a thread:

import asyncio
from muga import heartbeat

await asyncio.to_thread(heartbeat, "daily-cleanup")

Shutdown

Flush pending logs on graceful shutdown:

import signal
from muga import shutdown

def _handle(*_):
    shutdown()
    raise SystemExit(0)

signal.signal(signal.SIGTERM, _handle)

Configuration reference

Argument	Env var	Default
token	MUGA_TOKEN	(missing → SDK no-ops with a stderr warning)
endpoint	MUGA_ENDPOINT	https://api.muga.sh
service_name	MUGA_SERVICE	default
capture_exceptions	—	True
debug	—	False

init() also registers atexit.register(shutdown) on first successful call so any process exit drains queued records. Repeat calls do not double-register.

Reporting bugs

File a SDK bug report — the form prompts for SDK version, runtime, minimal repro, and expected vs actual. For anything potentially security-sensitive (token leaks, signature bypass, etc.), email [e-mail] instead of opening a public issue.