nova-fastapi-tool

A Hutool-style FastAPI utility toolkit.

Fast + Tool = nova-fastapi-tool. A curated, modular, batteries-included Python library that brings the elegance of Hutool (Java's beloved util library) to the FastAPI world.

Keep FastAPI sweet — one helper at a time.

✨ Features

🚀 Install

The package follows the "core + optional extras" pattern — exactly like Hutool's modular dependencies. Pick just what you need:

# Only core utilities (StrUtil / DateUtil / unified response / common middleware)
uv add nova-fastapi-tool

# + Nacos service registry & config center
uv add nova-fastapi-tool[nacos]

# + JWT auth / RBAC / password hashing / rate limiting
uv add nova-fastapi-tool[auth]

# + rich loguru-backed logging
uv add nova-fastapi-tool[logging]

# + OpenTelemetry 全链路追踪（Jaeger / 阿里云 ARMS / 自建 Collector）
uv add nova-fastapi-tool[tracing]

# + httpx-based HTTP client wrapper (HttpUtil / HttpClient)
uv add nova-fastapi-tool[http]

# + message queues — pick your broker (or [mq] for all four)
uv add nova-fastapi-tool[redis]      # Redis Streams (consumer groups)
uv add nova-fastapi-tool[rabbitmq]   # RabbitMQ (pika + aio-pika)
uv add nova-fastapi-tool[kafka]      # Kafka (kafka-python + aiokafka)
uv add nova-fastapi-tool[rocketmq]   # RocketMQ (official client)

# = Everything (≈ hutool-all)
uv add nova-fastapi-tool[all]

# = Dev / test / lint tooling
uv add nova-fastapi-tool[dev]

Python ≥ 3.11 is required.

uv add runs inside your own uv-managed project (the directory with pyproject.toml): it records the dependency and updates your uv.lock. uv is the only package manager.

🧩 5-second tour

from fastapi import FastAPI
from nova_fastapi_tool import create_app, LogUtil, R, StrUtil, DateUtil

# 1) Assemble a production-ready FastAPI app in 1 line
app: FastAPI = create_app(title="demo-service", debug=False, cors_origins=["*"])

# 2) Simple logging facade that behaves everywhere
LogUtil.info("Starting demo at {}", DateUtil.now_iso())

# 3) String helpers — Hutool-ish ergonomics
if StrUtil.is_blank("  "):
    LogUtil.warning("Empty input detected; masked={}", StrUtil.mask_email("user@example.com"))


# 4) Unified response envelope — your front-end will love you
@app.get("/hello")
def hello(name: str | None = None) -> R:
    return R.ok({"greeting": f"Hello, {StrUtil.or_default(name, 'World')}!"})

可观测性：Request ID 与 OpenTelemetry 双开关

Request ID 和 OpenTelemetry 是递进关系，不是互斥：两者使用不同的 Header （X-Request-ID vs traceparent）、存储不同的 ID（NanoId/UUID vs 32 位 hex trace_id），在 Header / 日志字段 / 上下文三个层面互不冲突，可同时开启。

create_app 把两个开关都接到位：RequestIdMiddleware 在外层先执行（生成或 提取 X-Request-ID，注入 LogUtil 上下文，并可选写入当前 OTel Span 的 attribute），FastAPI 自动埋点在内层创建 Server Span。开启后业务日志同时携带 request_id 与 trace_id——用 request_id 快速检索，用 trace_id 在 Jaeger / ARMS 看全链路。

from nova_fastapi_tool import create_app
from nova_fastapi_tool.logging import LogConfig

# 本地开发 / 单服务生产：只开 request_id（默认）
app = create_app(title="demo")

# 微服务 / 对接 ARMS：两个都开
app = create_app(
    title="pay-svc",
    log_config=LogConfig(
        tracing=True,
        otel_service_name="pay-svc",
        otel_exporter_endpoint="http://otel-collector:4317",  # 默认 localhost:4317
        otel_sample_rate=0.1,          # 采样 10%；默认 1.0 全量
        format_=True,                  # 日志输出扁平 JSON（SLS/ELK 友好）
    ),
)
# 或用 create_app 的快捷开关（覆盖 log_config.tracing）
app = create_app(title="pay-svc", tracing=True)

tracing=True 但未安装 [tracing] extra 时会抛出友好的 ImportError。 OTel 的 set_tracer_provider 是进程级一次性设置，重复调用只告警不覆盖。 完整设计文档见 docs/request-id-otel-dual-switch.md。

日志格式：扁平 JSON + 可注入格式

v0.5 起 loguru 后端在 json=True（即 format_=True）时输出扁平 JSON—— 不再是 loguru 自带的嵌套 {"text": …, "record": …}，而是每行一个 JSON 对象， 字段 service/timestamp/level/logger/message/path/line/func/thread_id/thread， 业务字段（默认 request_id、trace_id、span_id）提升到顶层，其余 extra 收进 extra 子对象——这正是 SLS / ELK / Loki 采集平台的期望结构。

from nova_fastapi_tool import LogUtil
from nova_fastapi_tool.logging import LogConfig

LogUtil.configure(
    LogConfig(
        json=True,                                  # 扁平 JSON
        top_level_fields=("request_id", "traceId", "task"),  # 提升到顶层的字段
        console_format="{time} | {level} | {message}",       # 覆盖控制台格式
        file_format_mode="json",                    # 文件固定 JSON（auto/text 可选）
        filter=lambda r: "health" not in r["message"],       # 用户过滤，可丢记录
    )
)
LogUtil.info("order created: {}", 1001)   # → {"message": "order created: 1001", …}

其他新配置项：file_format（文件格式独立覆盖）、console_format 在 stdlib 后端 接受 %-风格字符串或 logging.Formatter；opt(caller_depth=N) 暴露 loguru opt(depth=...) 等价能力，用于修正调用栈定位。

Quiet mode — mute third-party log noise

create_app already ships a structured AccessLogMiddleware (one rich line per request with method / path / status / duration / trace-id), so uvicorn's raw INFO: uvicorn.access access log is redundant — it is muted automatically and re-applied on server startup (uvicorn re-configures logging after the app module is imported):

app = create_app(title="demo-service")            # uvicorn.access muted by default
app = create_app(title="demo", mute_uvicorn_access=False)   # keep uvicorn's raw log
app = create_app(title="demo", access_log=False)  # no structured log → uvicorn's kept

For any other noisy stdlib logger (sqlalchemy.engine, …), use the generic switch on LogConfig (or call the helper directly):

from nova_fastapi_tool import LogUtil, mute_loggers
LogUtil.configure(mute_loggers=("uvicorn.access", "sqlalchemy.engine"))
mute_loggers("uvicorn.access")          # one-shot: clear handlers + disable propagation

Access log: request/response body capture + route-level overrides

AccessLogMiddleware is a pure-ASGI implementation (no BaseHTTPMiddleware overhead) emitting one structured line per request. Request and response bodies can be captured with independent switches and a unified byte cap — off by default, because body logging costs performance and carries PII risk:

from nova_fastapi_tool import AccessLogOptions, create_app

app = create_app(
    title="demo",
    router_prefix="/api",          # health/ready/info move to /api/health etc.
    access_log=AccessLogOptions(
        include_request_body=True,    # capture request bodies
        include_response_body=True,   # capture response bodies
        body_max_bytes=4096,          # unified safety cap (bytes)
    ),
)

Skip paths are split into two buckets: health_skip_paths (probe paths that move under a router prefix — create_app rewrites them to the full URL /api/health … once at build time, so they keep being skipped automatically) and skip_paths (app-root paths like /docs plus anything you add). No per-request prefix work; health probes stay skipped even when you override skip_paths.

Captured bodies are parsed by Content-Type instead of blind utf-8 decoding:

Sensitive query/body fields are masked as *** by default (token, password, api_key, secret, access_key, signature), and the list is configurable:

AccessLogOptions(sensitive_params=("token", "client_secret", "custom_key"))

Body capture can also be enabled per-route with the with_access_log decorator — the global middleware stays the single log emitter, so you get exactly one detailed line for that route and no duplicates:

from nova_fastapi_tool import with_access_log

@router.post("/compare")
@with_access_log(log_request_body=True, log_response_body=True)
async def compare(): ...

Migration (v1 → v2): the old single include_body switch is deprecated — it still works (sets both new switches) but emits a DeprecationWarning. Use include_request_body / include_response_body instead.

DB / SQL log output

Whether SQL statements are printed is a config-driven engine switch — never hardcoded in the framework. Use DbConfig (pydantic model, env-friendly) and hand its echo value to the engine at registration:

from sqlalchemy import create_engine
from nova_fastapi_tool import DbConfig, engine_registry

db_cfg = DbConfig.from_env()            # reads DB_ECHO / DB_ECHO_POOL (1/true/yes/on)
engine = create_engine("sqlite+pysqlite:///./app.db")
engine_registry.register_engine("default", engine, echo=db_cfg.echo)   # → engine.echo

• DB_ECHO=1 → every SQL statement + transaction boundary is printed (equivalent to create_engine(..., echo=True)); default off.
• DB_ECHO_POOL=1 → connection-pool activity.
• DbConfig also works inside your YAML-driven BaseSettings (plain pydantic nested model), and DbConfig.apply_to(engine) pushes both switches onto an existing engine at any time.
• register_engine(..., echo=None) (default) leaves the engine's own setting untouched.

With Nacos + JWT

See examples/ for:

• minimal_app.py — Logging + unified response skeleton
• tracing_demo.py — request_id + OpenTelemetry 双开关（扁平 JSON 日志）
• nacos_demo.py — Service register/discovery + config-center watcher
• auth_demo.py — JWT login + role-protected endpoints
• mq_demo.py — One MqUtil API across Redis / RabbitMQ / Kafka / RocketMQ

🧩 MQ in 30 seconds

from nova_fastapi_tool import MqUtil, RedisMQBackend  # or MemoryMQBackend / RabbitMQBackend / KafkaBackend / RocketMQBackend

mq = MqUtil(RedisMQBackend(url="redis://localhost:6379/0"))

# Publish any Python value — JSON codec is applied automatically
mq.publish("orders", {"order_id": 1001, "amount": 99.5}, key="user-7")

# One-shot poll (consumer-group semantics when `group` is given)
msg = mq.poll("orders", group="workers", timeout=5)
if msg:
    print(msg.payload)   # {"order_id": 1001, "amount": 99.5}
    mq.ack(msg)          # mark consumed; mq.nack(msg) redelivers

# Push consumer — handler receives a Message (`.payload` is already decoded);
# auto-acked on success, re-queued when the handler raises
sub = mq.subscribe("orders", lambda msg: print("handled", msg.payload))
...
sub.stop()
mq.close()

Async services use AsyncMqUtil + AsyncRedisMQBackend / AsyncRabbitMQBackend / AsyncKafkaBackend / AsyncRocketMQBackend — identical API, await everything. MemoryMQBackend needs no broker at all, so tests and local dev just work.

Redis has two modes behind the same RedisMQBackend:

• mode="streams" (default) — Streams + consumer groups: durable, ackable, competing consumers → real queue semantics.
• mode="pubsub" — Redis Pub/Sub broadcast: every live subscriber gets every message; nothing is persisted and poll is unsupported (push-only).

📁 Project layout

pypi-fastapi/
├── pyproject.toml           # PEP 621 packaging + extras (nacos/auth/logging/…)
├── src/
│   └── nova_fastapi_tool/            # Importable package (same name as PyPI artifact)
│       ├── __init__.py      # Re-exports everything (≈ hutool-all)
│       ├── version.py       # Single source of truth for version
│       ├── core/            # StrUtil, DateUtil, IdUtil, HashUtil, …
│       ├── logging/         # LogUtil facade + 扁平 JSON + RequestIdMiddleware + OTel 桥（tracing.py）
│       ├── config/          # Multi-profile YAML/.env + Settings base class
│       ├── nacos/           # NacosClient, Registry, ConfigCenter, Discovery, lifespan
│       ├── auth/            # JwtUtil, passwords, RBAC, rate-limit, Depends
│       ├── web/             # create_app, exceptions, R response, CORS, access-log MW
│       ├── db/              # DbUtil / AsyncDbUtil, Page, transactions, engine registry
│       ├── cache/           # CacheUtil / AsyncCacheUtil, memory/Redis backends
│       ├── http/            # HttpUtil / HttpClient, retry config, NovaHttpError
│       └── mq/              # MqUtil / AsyncMqUtil + Redis/RabbitMQ/Kafka/RocketMQ backends
├── tests/                   # Pytest suite, module-scoped
└── examples/                # Runnable demos

🛡️ License

Apache License 2.0 — see LICENSE.

📦 Publish to PyPI

PyPI credentials are kept out of this repo. Export a project-scoped API token as environment variables before uploading:

export UV_PUBLISH_USERNAME=__token__
export UV_PUBLISH_PASSWORD=pypi-xxxxxxxxxxxxxxxxxxxx
uv build               # → dist/nova_fastapi_tool-X.Y.Z{.tar.gz,.whl}
uv publish             # uploads dist/* (uv ≥ 0.5)