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 pip install nova-fastapi-tool

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

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

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

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

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

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

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

Python ≥ 3.11 is required.

Already working inside a uv-managed project? Use uv add nova-fastapi-tool[...] to add it as a project dependency (and update uv.lock) instead.

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

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)