"""Structured JSON logging for AgentForms.
Drop-in replacement for print() calls. Outputs JSON to stdout
with component, level, timestamp, and optional context fields.
Usage:
from app.services.logging import log
log.info("smtp", "Email sent", to=user_email)
log.error("webhook", "Delivery failed", error=str(e))
log.warning("rate", "Rate limit exceeded", ip=client_ip)
"""
import json
import logging
import os
import time
import uuid
from logging.handlers import RotatingFileHandler
logger = logging.getLogger("agentforms")
logger.setLevel(logging.DEBUG)
# Prevent duplicate handlers on re-import
if not logger.handlers:
# JSON formatter
class JsonFormatter(logging.Formatter):
def format(self, record):
entry = {
"ts": time.strftime("%Y-%m-%dT%H:%M:%S%z", time.gmtime(record.created)),
"level": record.levelname,
"component": getattr(record, "component", "-"),
"msg": record.getMessage(),
}
# Attach extra fields
extras = {}
for k, v in record.__dict__.items():
if k not in (
"name",
"msg",
"args",
"created",
"levelname",
"levelno",
"pathname",
"filename",
"module",
"funcName",
"lineno",
"exc_info",
"exc_text",
"stack_info",
"thread",
"threadName",
"processName",
"process",
"component",
"msecs",
"relativeCreated",
"taskName",
):
if v is not None:
extras[k] = v
if extras:
entry["ctx"] = extras
return json.dumps(entry)
# Console handler (stdout)
ch = logging.StreamHandler()
ch.setFormatter(JsonFormatter())
logger.addHandler(ch)
# File handler (rotating, in data/logs/)
log_dir = os.environ.get(
"LOG_DIR", os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "data", "logs")
)
os.makedirs(log_dir, exist_ok=True)
fh = RotatingFileHandler(
os.path.join(log_dir, "app.json"),
maxBytes=10 * 1024 * 1024, # 10MB
backupCount=3,
)
fh.setFormatter(JsonFormatter())
logger.addHandler(fh)
class StructuredLog:
"""Convenience wrapper: log.<level>(component, message, **ctx)"""
def debug(self, component: str, msg: str, **kwargs):
extra = {"component": component, **kwargs}
logger.debug(msg, extra=extra)
def info(self, component: str, msg: str, **kwargs):
extra = {"component": component, **kwargs}
logger.info(msg, extra=extra)
def warning(self, component: str, msg: str, **kwargs):
extra = {"component": component, **kwargs}
logger.warning(msg, extra=extra)
def error(self, component: str, msg: str, **kwargs):
extra = {"component": component, **kwargs}
logger.error(msg, extra=extra)
def critical(self, component: str, msg: str, **kwargs):
extra = {"component": component, **kwargs}
logger.critical(msg, extra=extra)
log = StructuredLog()
# Request ID tracking
_request_id = uuid.uuid4().hex[:8]
def get_request_id() -> str:
return _request_id