FastV — Laravel-style FastAPI Framework

一个基于 FastAPI 的全栈 Web 框架，把 Laravel 的架构范式（服务容器、门面、Eloquent 风格 ORM、Artisan 风格 CLI、中间件管道、任务队列、调度、广播）带到 Python，同时保留 FastAPI 的高性能与异步特性。

• 快速上手：安装 → 快速开始 → 目录结构
• 核心开发：配置 → 路由 → 控制器 → 模型 ORM → 验证 → 认证
• 基础设施：缓存 → Redis → 文件存储 → 日志
• 异步能力：队列 → 调度 → 事件 → WebSocket 广播
• 运维：中间件 → 异常处理 → CLI 命令 → 进程管理 → 连接池 → 健康检查

特性

• 服务容器（IoC） — 依赖注入与自动装配，支持 bind / singleton / instance / alias
• Eloquent 风格 ORM — 基于 SQLAlchemy 2.0，链式 QueryBuilder、软删除、分页/游标分页、类型转换、访问器/修改器、悲观锁、批量插入
• Artisan 风格 CLI — 代码生成器（make:model / make:controller / make:service / make:validate / make:channel）、数据库迁移、队列管理
• JWT 认证 — Guard、bcrypt 密码哈希（rounds 可配）、token 签发/校验、生产弱密钥强制
• 缓存系统 — Array / File / Redis 三驱动，键前缀、TTL、单飞（singleflight）、LRU、Redis 故障 fail-open 降级
• 任务队列 — AsyncTasQ 集成，重试、延迟任务、限流任务、批量分发
• 调度器 — APScheduler + Laravel 风格 DSL，防重叠、单实例执行、后台执行
• WebSocket / 广播 — Broadcast Facade、ShouldBroadcast 事件广播、public/private/presence 三型通道、JWT/Pusher 签名授权、Redis 跨进程桥，前端兼容 laravel-echo
• 文件存储 — 本地磁盘 + 阿里云 OSS（官方 v2 SDK）双驱动、命名磁盘、URL 生成
• 异步日志 — 队列后台写盘不阻塞、按日切分自动清理、JSON 结构化、request_id 上下文追踪、敏感信息脱敏、SQL/慢 SQL 日志、动态级别
• 统一异常处理 — 分层业务异常 + 统一 JSON 响应
• 健康检查 — /health（liveness）与 /health/ready（readiness，探活 DB/Redis + 连接池水位）

安装

# 核心框架
pip install fastv

# 一键安装全部可选依赖（MySQL/PostgreSQL/SQLite 驱动 + OSS）
pip install fastv[all]

可选 extras：dev（测试/ruff）、mysql（asyncmy）、postgres（asyncpg）、sqlite（aiosqlite）、oss（alibabacloud-oss-v2）。

要求：Python 3.12+，FastAPI 0.139+。

快速开始

# 1. 创建项目骨架（生成 app/、bootstrap/、config/、migrations/、.env）
mkdir myapp && cd myapp
fastv new

# 2. 编辑 .env 配置密钥（JWT_SECRET 生产必填 ≥32 字符）
# 3. 启动开发服务器（热重载 app/bootstrap/config + debug，就绪探针轮询 /health）
fastv start dev

# 4. 运行数据库迁移
fastv migrate

# 5. 生成模型/控制器/服务
fastv make:model User --migration
fastv make:controller UserController
fastv make:service UserService
fastv make:validate LoginRequest

目录结构

myapp/
├── app/
│   ├── controller/      # 控制器
│   ├── model/           # ORM 模型
│   ├── service/         # 业务逻辑层
│   ├── middleware/      # 中间件
│   ├── route/           # 路由文件
│   ├── validate/        # FormRequest 验证类
│   ├── queue/           # 队列任务定义
│   ├── command/         # 自定义 CLI 命令
│   └── channel/         # WebSocket 通道授权
├── bootstrap/           # 应用入口 main.py
├── config/              # 配置（app/database/redis/cache/queue/auth/
│                        #        storage/logging/websocket/middleware/exception）
├── migrations/          # Alembic 迁移
├── storage/             # 日志、缓存等运行时目录
├── .env                 # 环境变量（含密钥，勿提交）
└── alembic.ini

配置

所有配置在 config/*.py，参数通过 .env 覆盖（os.getenv）。点号路径访问：config.get("database.default")。

# config/database.py（模板生成，参数 .env 覆盖）
default = os.getenv("DB_CONNECTION", "sqlite")
connections = {
    "mysql": {
        "driver": "mysql",
        "driver_async": os.getenv("DB_ASYNC_DRIVER", "asyncmy"),  # asyncmy / aiomysql
        "host": os.getenv("DB_HOST", "127.0.0.1"),
        "pool_size": int(os.getenv("DB_POOL_SIZE", "10")),
        "max_overflow": int(os.getenv("DB_MAX_OVERFLOW", "20")),
        # ssl: {"ssl_ca": ...}  云数据库强制加密
    },
}

代码中读取配置：

from fastv import app as app_helper       # 或 app("service")
from fastv.support.helpers import config  # config("database.default")

db_default = config("database.default", "sqlite")

.env 关键项：APP_ENV（production 触发安全强制）、JWT_SECRET、DB_*、REDIS_*、CACHE_*、QUEUE_*、OSS_*、CORS_*、SECURITY_*、LOG_*。完整模板见 .env.example。

路由

路由文件放在 app/route/*.py，自动发现并注册。三种写法：

# app/route/web.py
from fastv.routing.router import Router
from app.controller.user import UserController

router = Router()

# 写法 1：方法引用（推荐，IDE 可跳转）
router.get("/users", UserController.index)

# 写法 2：Laravel 风格字符串
router.get("/users/{id}", "UserController@show")

# 写法 3：函数引用
async def health(request): ...
router.get("/health", health)

# 分组（闭包 / with / 链式）
with router.group(prefix="/admin", middleware=[AuthMiddleware]) as group:
    group.get("/dashboard", UserController.dashboard)

# 支持方法：get/post/put/patch/delete/any

路径参数、查询参数、请求体经 FastAPI 原生注入：

async def show(self, id: int, request):   # id 来自路径 {id}
async def store(self, request):            # request 自动注入 fastv Request

控制器

控制器继承 BaseController，自动注入当前请求，提供响应/异常快捷方法。

# app/controller/user.py
from fastv.controllers import BaseController

class UserController(BaseController):
    async def index(self):
        return self.success("ok", {"list": [...]})   # {code:1, message, data}

    async def show(self, id: int):
        user = await User.find(id)
        if not user:
            self.not_found("用户不存在")
        return self.json(user.to_dict())

    async def store(self):
        # 请求输入（JSON/form/query 合并，自动净化）
        name = await self.request.string("name")
        age = await self.request.integer("age")
        return self.success("created", data={"id": 1})

响应快捷方法：json(data, status)、success(message, data, **extra)、error(message, code, status)、paginate(paginator)、redirect(url)、html(content)、download(content, filename)。

异常快捷方法：throw / abort、fail、unauthorized、forbidden、not_found、server_error。

Request 便捷方法：all()、input(key)、string/integer/float/boolean/array（类型安全）、only(*keys)、except_(*keys)、file(key)、safe()（净化）、ip()（可信代理解析）、bearer_token()。

ResponseFactory（无控制器场景）：ResponseFactory.json/success/error/html/redirect，自动序列化 Model/datetime/date/Decimal/Enum/UUID。

模型

基于 SQLAlchemy 2.0 的 Laravel 风格 ORM。fastv new 后用 make:model 生成。

# app/model/user.py
from fastv.database import Model, SoftDeletes, TimestampsMixin
from fastv.database import fields as f
from sqlalchemy.orm import Mapped

class User(Model, SoftDeletes, TimestampsMixin):
    __tablename__ = "users"
    __fillable__ = ["name", "email", "password"]   # 批量赋值白名单
    __hidden__ = ["password"]                       # 序列化隐藏
    __casts__ = {"status": "int", "settings": "json"}  # 类型转换
    __appends__ = ["full_name"]                     # 追加虚拟属性

    id: Mapped[int] = f.pk()
    name: Mapped[str] = f.string(80)
    email: Mapped[str] = f.string(191)
    password: Mapped[str] = f.string(255)

    # 访问器：get_{name}_attribute
    def get_full_name_attribute(self): return f"{self.name} <{self.email}>"
    # 修改器：set_{name}_attribute（如密码自动哈希）
    def set_password_attribute(self, v): return hash_password(v)

字段工厂（fastv.database.fields）：pk()、int/bigint/small_int/tiny_int、string/text/medium_text、datetime/date、boolean（TINYINT(1)）。

查询（链式 QueryBuilder）

users = await User.query().where("status", 1).order_by("id", "desc").limit(10).get()
user = await User.query().where("email", email).first()
user = await User.find(1)
user = await User.find_or_fail(1)          # 不存在抛 NotFoundException
total = await User.query().where("status", 1).count()
ids = await User.query().pluck("id")

支持：where/or_where/where_in/where_null/where_between/when、order_by/order_desc/latest/oldest/in_random_order、group_by/having、limit/offset、select/distinct、with_（急加载）、paginate(page, per_page)、chunk/chunk_by_id、cursor_page（keyset 分页）、lock_for_update（悲观锁）。

# 分页
page = await User.query().paginate(1, 15)
# 游标分页（深分页性能好）
p1 = await User.query().cursor_page(column="id", per_page=20)
p2 = await User.query().cursor_page(column="id", last_cursor=p1.next_cursor)
# 悲观锁（事务内）
async with db.transaction() as s:
    order = await Order.query().where("id", 1).lock_for_update().first()

写入

user = await User.create({"name": "a", "email": "a@x.com"})
user.name = "b"; await user.save()
await user.update({"email": "new@x.com"})
await user.delete()
# 批量插入（executemany + 单事务，快一个数量级）
await User.insert_many([{"name": "a"}, {"name": "b"}], batch_size=500)

软删除

user = await User.find(1); await user.delete()   # 软删除（deleted_at）
await user.restore()                              # 恢复
await User.with_trashed().get()                   # 含软删除
await User.only_trashed().get()                   # 仅软删除
await user.force_delete()                         # 永久删除

关联

class Post(Model):
    def user(self): return self.belongs_to(User, foreign_key="user_id")
    # has_many / has_one / belongs_to

# 急加载（避免 N+1）
posts = await Post.query().with_("user").get()
post = posts[0]; post.relation("user")
# 条件关联
posts = await User.query().where_has("posts", lambda q: q.where("status", 1)).get()

死锁重试

from fastv.database import retry_on_deadlock

@retry_on_deadlock(retries=3)
async def decrement_stock(order_id: int):
    async with get_db().transaction() as s:
        order = await Order.query().where("id", order_id).lock_for_update().first()
        order.stock -= 1
        await order.save()

验证

FormRequest 基于 Pydantic，支持 Laravel 风格字符串规则与自定义规则。

# app/validate/login_request.py
from fastv.validation import FormRequest

class LoginRequest(FormRequest):
    @classmethod
    def rules(cls):
        return {
            "email": "required|email",
            "password": "required|min:8",
            "status": "in:pending,active",
        }

    @classmethod
    def messages(cls):
        return {"email.required": "邮箱不能为空"}

    @classmethod
    def only(cls):
        return ["email", "password"]   # 字段白名单（mass assignment 保护）

控制器中使用（自动从请求提取 + 净化 + 验证）：

async def login(self, req: LoginRequest):
    return self.success("ok", {"email": req.email})

自定义规则：

from fastv.validation import rule

rule("phone", lambda v: len(str(v)) == 11, "手机号格式不正确")

class RegisterRequest(FormRequest):
    rules = {"mobile": "required|phone"}

支持规则：required / optional / string / integer / boolean / email / url / min:N / max:N / in:a,b,c / 自定义。

认证

JWT 认证，bcrypt 密码哈希。

# 注册（AuthManager 自动从 config/auth.py 加载）
from fastv.auth import get_auth, AuthManager

# 登录
guard = get_auth().guard()
result = await guard.attempt({"email": email, "password": pwd})
if result:  # {"token": "...", "user": {...}}
    return self.success("login ok", result)

# 验证 token → 用户
user = await guard.validate_token(token)

# 密码哈希（自动升级旧算法，BCRYPT_ROUNDS 默认 12）
from fastv.auth import PasswordHasher
hasher = PasswordHasher()
hasher.make("plain"); hasher.check("plain", hashed)

生产环境 APP_ENV=production 时，JWT_SECRET 缺失/过短/默认值会拒绝启动。

缓存

三驱动（array/file/redis），默认 file。生产推荐 redis（含 fail-open 降级）。

from fastv.cache import CacheManager, get_cache

cache = get_cache()                        # 全局 CacheManager

# 直接操作
await cache.put("key", value, ttl=60)      # ttl 秒
value = await cache.get("key", default)
await cache.has("key"); await cache.delete("key")
await cache.increment("count", 2)

# remember（缓存穿透防护 + 单飞 singleflight）
result = await cache.remember("report", ttl=300, callback=expensive_fn)

# 命名存储
file_store = cache.store("file")

config/cache.py：stores 支持 array（可配 max_items LRU）、file（path）、redis（connection）。Redis 故障时自动降级 array（不拖垮业务）。

Redis

完整 Redis 客户端（ConnectionPool 共享 + 单例 + Facade），生产级：连接验证、故障快速失败。

from fastv.redis import Redis, get_redis

# Facade（类方法，异步）
await Redis.set("key", "value", ex=60)
value = await Redis.get("key")
await Redis.hset("user:1", "name", "张三")

# 命名连接
await Redis["cache"].get("key")

# 直接使用连接
conn = get_redis().connection("default")
await conn.lock("order:123", timeout=10)     # 分布式锁
async with conn.lock("order:123") as lock: ...
await conn.publish("channel", "msg")
async for msg in conn.pubsub().listen(): ...
await conn.rate_limit("api:user:1", max_calls=100, window=60)

支持的完整数据类型：String / Hash / List / Set / ZSet / HyperLogLog / Stream / PubSub / 分布式锁 / 事务 Pipeline / Scan。Redis 故障时首次操作 1s 内快速失败并标记未连接，后续瞬时返回（fail-fast）。

文件存储

本地磁盘 + 阿里云 OSS（官方 v2 SDK）双驱动。

from fastv.storage import Storage

# 本地（默认）
await Storage.put("uploads/a.txt", "hello")
content = await Storage.get("uploads/a.txt")
url = Storage.url("uploads/a.txt")           # /storage/uploads/a.txt
await Storage.delete("uploads/a.txt")
files = await Storage.files("uploads")       # 列表

# OSS（命名磁盘）
await Storage["oss"].put("uploads/a.txt", "hello")
url = Storage["oss"].url("uploads/a.txt")    # https://bucket.endpoint/prefix/a.txt

config/storage.py：

disks = {
    "local": {"driver": "local", "root": os.getenv("STORAGE_LOCAL_ROOT", "public"), "url": "/storage"},
    "oss": {
        "driver": "oss",
        "access_key_id": os.getenv("OSS_ACCESS_KEY_ID"),
        "access_key_secret": os.getenv("OSS_ACCESS_KEY_SECRET"),
        "endpoint": os.getenv("OSS_ENDPOINT"),        # oss-cn-heyuan.aliyuncs.com
        "bucket": os.getenv("OSS_BUCKET_NAME"),
        "prefix": os.getenv("OSS_PREFIX", ""),
        "url": os.getenv("OSS_URL", ""),              # 自定义域名/CDN
        "region": os.getenv("OSS_REGION", ""),        # 可选，自动从 endpoint 推断
    },
}

日志

异步队列写盘不阻塞事件循环；按日切分 + 自动清理；JSON 结构化可选。

from fastv.log import Log

Log.info("User logged in")
Log.error("Something failed", exc_info=True)
Log.log_queue("job done")                       # 队列专用日志
Log.set_level("error")                          # 动态级别

# 上下文注入（日志携带 user_id/trace_id）
token = Log.with_context(user_id=123, trace_id="tr-1")
try:
    Log.info("processing order")
finally:
    from fastv.log.context import reset_context
    reset_context(token)

config/logging.py：

• async_logging：队列后台线程写盘（默认 true）
• channels：daily（默认）/ single / json（结构化）/ stderr / stack（组合）
• sql：SQL 日志开关 + 慢 SQL 阈值 slow_ms（超时记录到 slow-sql.log）

特性：request_id 跨线程正确、敏感信息脱敏（password/token/Bearer）、SQL/慢 SQL 统一落盘。

队列

基于 AsyncTasQ 的任务队列。

# app/queue/__init__.py 或任务文件
from fastv.queue import Task, Queue, Bus

class SendEmail(Task):
    queue = "emails"

    async def handle(self):
        to = self.payload.get("to")
        # ... 发送逻辑

# 分发单个任务
await SendEmail(to="user@test.com").dispatch()

# 延迟 / 指定队列
await SendEmail(to="x").delay(60).dispatch()
await SendEmail(to="x").on_queue("high").dispatch()

# 批量分发
task_ids = await Bus.dispatch([SendEmail(a), SendEmail(b)])

# 统计
stats = await Queue.stats()
size = await Queue.size("emails")

启动 Worker：

fastv queue:work --queue=default,emails --concurrency=10

Worker 连接池自动随并发自适应（pool_size = max(配置, concurrency)），优雅关闭清理数据库/队列/残留任务。

限速任务（上游 API 限流场景）

调用有速率上限的第三方 API（如生图接口）时，用 RateLimitedTask 控制消费速率。任务在 handle() 里排队等待配额（Redis 固定窗口计数器），窗口释放后自动继续，不烧重试次数。

# app/queue/image_task.py
from fastv.queue import RateLimitedTask

class GenerateImageTask(RateLimitedTask):
    queue = "images"
    max_retry_attempts = 3
    retry_delay_seconds = 30

    # ---- 限速配置（对齐上游 API 上限）----
    rate_limit = 10            # 窗口内最多执行 10 次
    rate_limit_window = 60     # 窗口 60 秒 → 10 QPM（每分钟 10 张）
    rate_limit_timeout = 300   # 排队等配额最多 5 分钟，超时抛异常按 retry_delay 重试
    rate_limit_poll_interval = 1.0  # 每 1 秒轮询检查一次配额
    timeout_seconds = 120      # 单张生成超时 120s

    def __init__(self, prompt: str, model: str = "sd-xl"):
        super().__init__(prompt=prompt, model=model)

    async def execute(self):
        result = await image_api.generate(self.prompt, self.model)
        return result

# 分发
await GenerateImageTask(prompt="一只猫", model="sd-xl").dispatch()

限速参数换算：

按用户/渠道隔离限速（每个用户独立计数）：

class GenerateImageTask(RateLimitedTask):
    rate_limit = 5
    rate_limit_window = 60

    def __init__(self, user_id: int, prompt: str):
        super().__init__(user_id=user_id, prompt=prompt)

    @property
    def rate_limit_key(self) -> str:
        return f"user:{self.user_id}"   # 每个用户独立 5 QPM

对比：主动抛异常重试（Task + RateLimitExceeded）：

from fastv.queue.rate_limiter import get_queue_rate_limiter, RateLimitExceeded

limiter = get_queue_rate_limiter()
allowed = await limiter.acquire("sms", max_calls=100, window=60)
if not allowed:
    raise RateLimitExceeded(retry_after=60)   # 按 retry_delay_seconds 重试

生图等上游限流场景推荐 RateLimitedTask（排队等待，避免"重试→再限流"死循环）；需要延迟后重试的场景用主动抛异常。

调度

APScheduler + Laravel 风格 DSL。

# app/schedule.py
from fastv.schedule import scheduler

scheduler.command("cache:clear").daily_at("02:00")
scheduler.call(sync_data, "0 */6 * * *")
scheduler.every(5).minutes().call(send_digest).without_overlapping()

# 执行语义控制
scheduler.call(expensive_job, "*/10 * * * *").without_overlapping()   # 防重叠
scheduler.call(report, "0 8 * * *").on_one_server()                   # 单实例（Redis 锁）
scheduler.call(bg_task, "*/5 * * * *").run_in_background()            # 后台执行

DSL：cron / every(n).minutes/seconds/hours/days / every_minute / hourly_at / daily_at("HH:MM") / weekly / monthly / mondays().at("08:00") 等。

启动守护进程：

fastv schedule:serve

on_one_server() 基于 Redis 锁（SET NX EX），多 worker 部署时保证同一定时任务只执行一次；without_overlapping() 任务运行期间跳过重叠触发。

事件

发布/订阅，支持通配符与监听器优先级。

from fastv import app

dispatcher = app("events")   # 或 app.make("events")

# 注册（可在 app/event/*.py）
dispatcher.listen("user.created", handle_user_created)
dispatcher.listen("user.*", handle_all_user, priority=10)

# 触发（推荐 async）
await dispatcher.dispatch_async("user.created", user_data)
# 或同步（事件循环内异步监听器自动转后台任务）
dispatcher.dispatch("user.created", user_data)

# 通配符 / 优先级 / 清空
dispatcher.has_listeners("user.created")
dispatcher.forget("user.created"); dispatcher.flush()

# 等待后台异步监听器完成（优雅关闭/测试）
failed = await dispatcher.await_pending(timeout=5)

事件监听器也可用类（实现 subscribe(dispatcher)）自动发现于 app/event/*.py。

WebSocket 广播

类 Laravel Broadcasting + Reverb，前端兼容 laravel-echo。

后端广播

# 1. 定义事件实现 ShouldBroadcast
from fastv.websocket import ShouldBroadcast

class OrderShipped(ShouldBroadcast):
    broadcast_on = ["private-orders.1"]
    broadcast_as = "order.shipped"
    broadcast_with = {"order_id": 1}

# 2. 直接广播
from fastv.websocket import Broadcast
await Broadcast.dispatch(OrderShipped())

# 或经事件系统自动广播
await dispatcher.dispatch_async("order.shipped", OrderShipped())

# 3. 原始通道广播
await Broadcast.channel("public-news", "news.created", {"id": 1})

通道授权

# app/channel/order.py
from fastv.websocket import channel

channel("private-order.{orderId}", lambda user, orderId: user.id == int(orderId))
channel("presence-room.{roomId}", authorize)

async def authorize(user, roomId):
    return user is not None and roomId in user.rooms

channel(name, callback)：{param} 占位匹配路径参数，user 为 JWT 解析用户（未认证 None），返回 truthy 允许订阅。

前端接入（laravel-echo）

import Echo from 'laravel-echo';
import Pusher from 'pusher-js';

window.Echo = new Echo({
    broadcaster: 'pusher', key: 'fastv',
    wsHost: window.location.hostname, wsPort: 8000,
    forceTLS: false, disableStats: true,
    auth: { headers: { Authorization: 'Bearer <jwt>' } },
});

Echo.private('orders.1').listen('order.shipped', e => console.log(e));

配置：config/websocket.py（enabled / path=/ws / driver=memory|redis / app_secret）。driver=redis 时跨 worker/实例广播（Redis Pub/Sub 桥）。

中间件

框架内置安全中间件（参数 .env）+ 自定义中间件自动发现。

# app/middleware/auth.py（自动发现：有 __call__ 或 handle 的类）
class AuthMiddleware:
    async def handle(self, request, call_next):
        if not request.state.user:
            return JSONResponse({"code": -1, "message": "未登录"}, status_code=401)
        return await call_next(request)

内置中间件（.env 控制）：

• RequestIdMiddleware — X-Request-ID + 日志 contextvar（默认启用）
• RequestBodyLimitMiddleware — 请求体大小限制（REQUEST_BODY_LIMIT，默认 10MB）
• SecurityHeadersMiddleware — nosniff/frame/referrer/CSP（SECURITY_*）
• AccessLogMiddleware — 访问日志（ACCESS_LOG_ENABLED）
• CORSMiddleware — CORS（CORS_*，默认关闭）
• ThrottleMiddleware — 限速（Redis 滑动窗口，TRUSTED_PROXIES 白名单防伪造 IP）

限速（config/middleware.py 注册）：

global_middleware = [
    "fastv.http.middleware.throttle.ThrottleMiddleware",
]

异常处理

统一返回 {code, message, data}。config/exception.py 可配置消息/状态码覆盖。

from fastv.exceptions import (
    BusinessException, UnauthorizedException, ForbiddenException,
    NotFoundException, ServerException, FrameworkException,
)

raise NotFoundException("用户不存在")
raise BusinessException(message="库存不足", data={"id": 1})

# 自定义异常（有 code/http_status/message 属性即被识别）
class MyError(Exception):
    code = -100; http_status = 400
    def __init__(self, message): self.message = message

CLI 命令

fastv new                  # 创建项目骨架
fastv key:generate         # 生成 JWT_SECRET 并写入 .env（--show 仅打印，--force 覆盖）
fastv start dev            # 开发服务器（热重载 app/bootstrap/config）
fastv start prod           # 生产（多 worker）
fastv serve                # HTTP + 队列 worker 一起
fastv migrate              # 运行迁移
fastv make:model User --migration
fastv make:controller UserController --resource
fastv make:service UserService
fastv make:validate LoginRequest
fastv make:channel Order
fastv make:command SendReport
fastv queue:work --queue=default --concurrency=10
fastv queue:stats
fastv schedule:serve        # 定时任务守护进程
fastv cache:clear

自定义命令（app/command/*.py 自动发现）：

from fastv.console.command import Command

class SendReportCommand(Command):
    signature = "send:report {--force}"
    description = "发送日报"

    def handle(self):
        force = self.option("force")
        self.info(f"发送日报 force={force}")
        return 0

进程管理

• fastv start dev/prod — 子进程 uvicorn，独立进程组，Ctrl+C 优雅关闭（超时强杀进程树），启动后就绪探针
• fastv serve — uvicorn + 队列 worker 双进程，信号转发与互监控
• fastv queue:work — 队列 Worker，跨平台信号（Windows CTRL_BREAK / POSIX SIGTERM），优雅清理
• fastv schedule:serve — 调度守护进程，显式信号 → 优雅关闭

生产部署建议 systemd（Restart=always）守护 fastv start prod。

连接池

MySQL / Redis 连接池是进程内资源，各进程独立池（共享配置与 Redis 服务器）。

• 进程内（uvicorn worker）：WebSocket/cache/限速/队列分发共享同一 engine 与 Redis pool
• 跨进程：多 worker / queue:work / schedule:serve 各自建池

建议池大小：worker 5~10、队列 worker 自动随并发自适应、调度 2~3。

⚠️ 多 worker 注意连接总量：默认峰值 30/进程 × 6 进程 = 180 > MySQL 默认 max_connections=151。

健康检查

GET /health           # liveness：status + version/uptime/env
GET /health/ready     # readiness：DB/Redis 探活 + database.pool 连接池水位

/health/ready 返回 {status, checks: {database: {status, pool}, redis: {status}}, ...}，依赖故障返回 503。

安全清单（生产上线前）

• .env 不入 git（已配置），轮换历史泄露的 JWT_SECRET / DB_PASSWORD / OSS AK
• JWT_SECRET ≥32 字符随机串（生产弱密钥会拒绝启动）
• APP_ENV=production、APP_DEBUG=false
• MySQL 连接池总量 ≤ max_connections
• OSS/Redis 密码使用强凭证
• 若需反向代理，配置 TRUSTED_PROXIES（否则限速/日志不信任代理头）
• 需要跨域时显式配置 CORS_ORIGINS（默认关闭）

许可证

MIT