Best for
- Setting up Python logging for any service or script
- Configuring structured JSONL logging for analysis
- Implementing log rotation
terrylica/cc-skills/plugins/devops-tools/skills/python-logging-best-practices/SKILL.md
Python logging with loguru, structlog, and orjson. TRIGGERS - loguru, structlog, structured logging
Decision brief
Self-Evolving Skill: This skill improves through use. If instructions are wrong, parameters drifted, or a workaround was needed — fix this file immediately, don't defer. Only update for real, reproducible issues.
Compatibility matrix
| Platform | Status | Evidence | What to check |
|---|---|---|---|
| Codex | Not declared | No explicit evidence | Portability before use |
| Claude Code | Declared | Source record | Install path and trigger |
| Cursor | Not declared | No explicit evidence | Portability before use |
| Gemini CLI | Not declared | No explicit evidence | Portability before use |
Installation
The source command is displayed only when detected. A safe inspection prompt is always available so your agent can explain every action before execution.
npx skills add https://github.com/terrylica/cc-skills --skill "plugins/devops-tools/skills/python-logging-best-practices"Inspect the Agent Skill "python-logging-best-practices" from https://github.com/terrylica/cc-skills/blob/a5f847b22ee5afa35677e446973a903d098cd1d4/plugins/devops-tools/skills/python-logging-best-practices/SKILL.md at commit a5f847b22ee5afa35677e446973a903d098cd1d4. List every install step, command, network request, credential, file read/write, external action, and rollback step. Explain whether it fits my task. Do not install or execute anything until I approve.
Workflow
logevent("tokenrefresh", {"account": name, "tokenfp": tokenfingerprint(token)}) python @app.get("/api/status") def status(): """White-box monitoring — current state on demand.""" return {"activeaccount": ..., "accounts": [...], "polledat": ...}
Setting up Python logging for any service or script
Review the “Decision Heuristic: Start Light, Scale Up” section in the pinned source before continuing.
For: < 5 systemd services, single server, single operator. Battle-tested in production by ccmax-monitor.
Review the “Architecture: Three-Concern Separation” section in the pinned source before continuing.
Permission review
The documentation asks the agent to create, modify, or delete local files.
print(f"[telemetry] write failed: {e}", file=__import__("sys").stderr, flush=True)The documentation asks the agent to create, modify, or delete local files.
dst.unlink(missing_ok=True)Evidence record
| Signal | Value | Evidence type | Meaning |
|---|---|---|---|
| Quality score | 95/100 | Computed | Documentation, specificity, maintenance, and trust rules |
| Repository stars | 62 | Source | Repository attention, not individual Skill quality |
| Compatibility | 1 platforms | Source | Declared in the catalog source record |
| Usage guide | automated source guide | Editorial | Generated or reviewed according to the visible evidence level |
Pinned source
Self-Evolving Skill: This skill improves through use. If instructions are wrong, parameters drifted, or a workaround was needed — fix this file immediately, don't defer. Only update for real, reproducible issues.
Use this skill when:
Unified reference for Python logging patterns optimized for machine readability (Claude Code analysis) and operational reliability. Starts with the lightest viable approach and scales up only when needed.
Is it < 5 services on a single machine, < 1 event/sec?
YES → Lightweight Pattern (print + JSONL telemetry)
NO → Is it containerized / serverless?
YES → stdout JSON (any library), no file rotation
NO → Is OTel tracing required?
YES → structlog + OTel
NO → loguru (CLI tools) or stdlib RotatingFileHandler
| Approach | Use Case | Pros | Cons |
|---|---|---|---|
| Lightweight | Small systemd services, self-hosted, single operator | Zero deps, journald integration, minimal code | No severity filtering, no per-module control |
loguru | CLI tools, scripts, local services | Zero-config, built-in rotation, great DX | External dep, not truly schema-enforced |
structlog | Production services, OTel integration | ContextVars, processor chains, OTel-native | Steeper learning curve |
stdlib | LaunchAgent daemons, zero-dep constraint | No dependencies, Python 3.14 merge_extra | More boilerplate, no structured defaults |
Logfire | AI/LLM observability, Pydantic apps | Built on OTel, token/cost tracking, SQL | SaaS dependency, newer ecosystem |
For: < 5 systemd services, single server, single operator. Battle-tested in production by ccmax-monitor.
This pattern uses a two-channel architecture:
print(flush=True) → systemd journald (operational logs, human-readable)This maps to the 12-Factor App's "treat logs as event streams" principle. journald handles ops (rotation, filtering, metadata), while the JSONL file serves domain telemetry for post-mortem analysis.
| Concern | Mechanism | Purpose | Lifecycle |
|---|---|---|---|
| Ops logging | print() → journald | Human debugging, journalctl -u service -f | Managed by journald (auto-rotated) |
| Telemetry | JSONL file (telemetry.jsonl) | Structured audit trail, AI/LLM analysis | Append-only, rotated by size |
| State recovery | WAL file (optional) | Crash recovery for irreversible operations | Ephemeral, deleted on success |
"""Append-only JSONL telemetry logger with size-based rotation.
Zero external dependencies. Works with systemd journald for ops logging
and a separate JSONL file for structured machine-readable telemetry.
"""
import json
from datetime import datetime, timezone
from pathlib import Path
TELEMETRY_PATH = Path(__file__).parent / "telemetry.jsonl"
MAX_SIZE = 10 * 1024 * 1024 # 10 MB
BACKUP_COUNT = 3 # Keep 3 rotated backups (~30MB total)
def log_event(event_type: str, data: dict) -> None:
"""Append a structured JSON line to telemetry.jsonl."""
entry = {
"ts": datetime.now(timezone.utc).isoformat(),
"type": event_type,
**data,
}
line = json.dumps(entry, separators=(",", ":")) + "\n"
try:
try:
if TELEMETRY_PATH.stat().st_size > MAX_SIZE:
_rotate()
except FileNotFoundError:
pass
with open(TELEMETRY_PATH, "a") as f:
f.write(line)
except OSError as e:
# Fallback to stderr (captured by journald)
print(f"[telemetry] write failed: {e}", file=__import__("sys").stderr, flush=True)
def _rotate() -> None:
"""Rotate telemetry files: .jsonl → .jsonl.1 → .jsonl.2 → .jsonl.3"""
for i in range(BACKUP_COUNT, 1, -1):
src = TELEMETRY_PATH.with_suffix(f".jsonl.{i - 1}")
dst = TELEMETRY_PATH.with_suffix(f".jsonl.{i}")
if src.exists():
dst.unlink(missing_ok=True)
src.rename(dst)
backup = TELEMETRY_PATH.with_suffix(".jsonl.1")
backup.unlink(missing_ok=True)
TELEMETRY_PATH.rename(backup)
# === Ops logging (goes to journald via stdout) ===
def log(msg: str) -> None:
"""Human-readable operational log line. Captured by journald."""
ts = datetime.now(timezone.utc).strftime("%H:%M:%S")
print(f"[{ts}] {msg}", flush=True)
Usage:
# Operational (human reads via journalctl -u myservice -f)
log("Refreshing token for account X")
log("Switch: account A → account B (reason: 5h breach)")
# Telemetry (machine reads via jq/DuckDB/Claude Code)
log_event("token_refresh", {"account": "X", "expires_in_h": 8.0, "token_fp": "abc12345"})
log_event("account_switch", {"from": "A", "to": "B", "reason": "5h_breach"})
Never pass secrets through the logging pipeline. Log only a non-reversible fragment:
def _token_fingerprint(token: str) -> str:
"""Extract uniquely identifiable chars from a token's mid-section.
The prefix (sk-ant-oat01-) and suffix (...AA) are common across tokens.
Chars 14-22 (after the prefix) are the most unique per-token.
Middle-slice avoids leaking type-prefix metadata that prefix-based
approaches expose.
"""
if len(token) > 25:
return token[14:22]
return token[:8] if token else ""
# Usage: log the fingerprint, never the token
log_event("token_refresh", {"account": name, "token_fp": _token_fingerprint(token)})
Why this is superior to regex redaction filters:
| Approach | Security | Maintenance | Failure mode |
|---|---|---|---|
| Token fingerprinting (log only a slice) | Secret never enters logging pipeline | Zero — works with any token format | Cannot fail — nothing to redact |
| Regex redaction filter | Secret passes through, filtered on output | Must update regexes for new token formats | Silent miss = secret in logs |
This aligns with OWASP Logging Cheat Sheet: "Ensure that no sensitive data is included in log entries." Major platforms (AWS, Stripe, GitHub) use separate non-secret identifiers or partial token display — never full tokens with regex scrubbing.
Regex filters remain useful as a defense-in-depth backstop, not a primary control.
For small deployments, rich JSON health endpoints replace log aggregation:
@app.get("/api/status")
def status():
"""White-box monitoring — current state on demand."""
return {"active_account": ..., "accounts": [...], "polled_at": ...}
@app.get("/api/vault-health")
def vault_health():
"""Token health for all accounts."""
return {name: {"status": "healthy", "expires_in": "7.5h", ...} for ...}
This is the Health Endpoint Monitoring Pattern (Microsoft Azure Architecture Center) / Health Check API Pattern (microservices.io). The dashboard IS the monitoring tool — no Grafana/Prometheus needed.
When the service itself serves its own operational state as structured JSON, you get:
curl and analyze directly)No log aggregation stack needed. These single-binary tools work directly on JSONL:
# DuckDB — SQL analytics on JSONL (most powerful)
duckdb -c "SELECT type, count(*) FROM read_json_auto('telemetry.jsonl') GROUP BY 1 ORDER BY 2 DESC"
# jq — ad-hoc JSON filtering
jq 'select(.type == "token_refresh")' telemetry.jsonl
# journalctl — already exports JSONL natively
journalctl -u ccmax-switcher -o json --since "1h ago" | jq 'select(.PRIORITY == "3")'
# lnav — interactive terminal log viewer with SQL
lnav telemetry.jsonl
# llm (Simon Willison) — pipe to LLM for AI post-mortem
journalctl -u myservice --since "2h ago" --priority=err -o json | llm "analyze root cause"
Upgrade to loguru/structlog when any of these become true:
orjson)For CLI tools, scripts, and services that benefit from a logging library:
from loguru import logger
logger.add(
log_path,
rotation="10 MB",
retention="7 days",
compression="gz"
)
# stdlib alternative (zero-dep)
from logging.handlers import RotatingFileHandler
handler = RotatingFileHandler(
log_path,
maxBytes=100 * 1024 * 1024, # 100MB
backupCount=5
)
Container/serverless apps: Skip file rotation entirely. Log to stdout/stderr as JSON. Let the container runtime handle collection and rotation.
# One JSON object per line - jq-parseable
{"timestamp": "2026-01-14T12:45:23.456Z", "level": "info", "message": "..."}
File extension: Always use .jsonl (not .json or .log)
Performance: For >10k records/sec, use orjson instead of json.dumps():
import orjson
def json_formatter(record) -> str:
log_entry = { ... }
return orjson.dumps(log_entry).decode()
Use as a backstop alongside token fingerprinting, not as the primary control:
import re
REDACT_PATTERNS = [
(re.compile(r'AKIA[0-9A-Z]{16}'), '[REDACTED_AWS_KEY]'),
(re.compile(r'sk-[a-zA-Z0-9]{48}'), '[REDACTED_API_KEY]'),
(re.compile(r'(?i)bearer\s+[a-zA-Z0-9._~+/=-]+'), '[REDACTED_BEARER]'),
]
def redact_filter(record):
for pattern, replacement in REDACT_PATTERNS:
record["message"] = pattern.sub(replacement, record["message"])
return True
logger.add(sink, filter=redact_filter)
import asyncio
from loguru import logger
async def main():
logger.add("app.jsonl", enqueue=True)
await logger.complete()
asyncio.run(main())
# Sync: logger.remove()
#!/usr/bin/env python3
# /// script
# requires-python = ">=3.14"
# dependencies = ["loguru", "orjson"]
# ///
import re
import sys
from pathlib import Path
from uuid import uuid4
import orjson
from loguru import logger
REDACT_PATTERNS = [
(re.compile(r'AKIA[0-9A-Z]{16}'), '[REDACTED_AWS_KEY]'),
(re.compile(r'sk-[a-zA-Z0-9]{48}'), '[REDACTED_API_KEY]'),
]
def json_formatter(record) -> str:
log_entry = {
"timestamp": record["time"].strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
"level": record["level"].name.lower(),
"component": record["function"],
"operation": record["extra"].get("operation", "unknown"),
"operation_status": record["extra"].get("status", None),
"trace_id": record["extra"].get("trace_id"),
"message": record["message"],
"context": {k: v for k, v in record["extra"].items()
if k not in ("operation", "status", "trace_id", "metrics")},
"metrics": record["extra"].get("metrics", {}),
"error": None
}
if record["exception"]:
exc_type, exc_value, _ = record["exception"]
log_entry["error"] = {
"type": exc_type.__name__ if exc_type else "Unknown",
"message": str(exc_value) if exc_value else "Unknown error",
}
return orjson.dumps(log_entry).decode()
def redact_filter(record):
for pattern, replacement in REDACT_PATTERNS:
record["message"] = pattern.sub(replacement, record["message"])
return True
def setup_logger(app_name: str, log_dir: Path | None = None):
logger.remove()
logger.add(sys.stderr, format=json_formatter, filter=redact_filter, level="INFO")
if log_dir is not None:
log_dir.mkdir(parents=True, exist_ok=True)
logger.add(
str(log_dir / f"{app_name}.jsonl"),
format=json_formatter,
filter=redact_filter,
rotation="10 MB",
retention="7 days",
compression="gz",
level="DEBUG"
)
return logger
| Field | Type | Purpose |
|---|---|---|
timestamp / ts | ISO 8601 | Event ordering (millisecond precision minimum) |
level / type | string | Severity or event type |
component / svc | string | Module, function, or service name |
operation | string | What action is being performed |
operation_status | string | started/success/failed/skipped |
trace_id | UUID4 or OTel | Correlation ID (OTel trace ID for production services) |
message | string | Human-readable description |
context | object | Operation-specific metadata |
metrics | object | Quantitative data (counts, durations) |
error | object/null | Exception details if failed |
enqueue=True without logger.complete() - Silent log loss on shutdownenqueue=True with slow sinks - Unbounded memory growthjson.dumps() at >10k events/sec - Use orjson for 2-10x speedup| Issue | Cause | Solution |
|---|---|---|
| loguru not found | Not installed | Run uv add loguru |
| Logs not appearing | Wrong log level | Set level to DEBUG for troubleshooting |
| Log rotation not working | Missing rotation config | Add rotation param to logger.add() |
| JSONL parse errors | Malformed log line | Check for unescaped special characters |
| OOM with enqueue=True | Unbounded internal queue | Monitor RSS; use structlog or avoid slow sinks |
| Lost logs on shutdown | Missing logger.complete() | Call await logger.complete() or logger.remove() |
| Slow JSONL serialization | Using stdlib json at high volume | Switch to orjson.dumps().decode() |
| Secrets in logs | No fingerprinting | Log token slices, not full values |
| journald not capturing output | Missing flush | Use print(..., flush=True) or PYTHONUNBUFFERED=1 |
| No alerts when services crash | No external monitor | Add Uptime Kuma or Gatus polling health endpoints |
After this skill completes, check before closing:
Only update if the issue is real and reproducible — not speculative.
Frequently asked questions
Self-Evolving Skill: This skill improves through use. If instructions are wrong, parameters drifted, or a workaround was needed — fix this file immediately, don't defer. Only update for real, reproducible issues.
The source record exposes this install command: npx skills add https://github.com/terrylica/cc-skills --skill "plugins/devops-tools/skills/python-logging-best-practices". Inspect the command and pinned source before running it.
The pinned source record declares support for: claude code.
Static rules flagged write-files in the source; the page lists the matching lines and excerpts.
Alternatives
ljagiello/ctf-skills
Provides binary exploitation techniques for CTF challenges. Use when you already have a vulnerable native target or service and need to turn memory corruption or low-level primitives into code execution or privilege escalation, such as buffer overflows, format strings, heap bugs, ROP, ret2libc, shellcode, kernel exploitation, seccomp bypass, sandbox escape, or Windows/Linux exploit chains. Do not use it when the main blocker is understanding what the binary does; use reverse engineering first. D
PramodDutta/qaskills
Test automation with Gauge framework using Markdown specifications, step implementations in Java/Python/JavaScript/Ruby/C#, concepts, data-driven testing, and living documentation.
K-Dense-AI/scientific-agent-skills
Distributed computing for larger-than-RAM pandas/NumPy workflows. Use when you need to scale existing pandas/NumPy code beyond memory or across clusters. Best for parallel file processing, distributed ML, integration with existing pandas code. For out-of-core analytics on single machine use vaex; for in-memory speed use polars.
rampstackco/claude-skills
Running experiments out of the data warehouse instead of via dedicated experiment platforms. SQL-based assignment, exposure logging discipline, metric definitions in dbt models, statistical analysis in SQL or Python, variance reduction with CUPED, sequential testing, and the operational tradeoffs vs platforms like Statsig and Optimizely. Triggers on warehouse-native experimentation, run experiments in BigQuery, run experiments in Snowflake, dbt experiments, SQL t-test, CUPED variance reduction,