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]

# + 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')}!"})

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

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
• 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 formatter + request-ID MW
│       ├── 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)