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）、permission（Casbin 授权），以及把运行时可选依赖一次装齐的 all。

要求：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

项目代码怎么被导入：app 由框架注册成一个包对象（sys.modules["app"] 指向本项目的 app/）， 项目里一律写 from app.controller.user import UserController 这种带 app. 前缀的绝对导入， 包内也可以用相对导入（from .user import X）。

框架不把项目路径放进 sys.path：一旦放进去，app/ 下与标准库重名的目录就会顶掉标准库 （脚手架生成的 app/queue/ 顶掉标准库 queue，启动即 AttributeError: module 'queue' has no attribute 'Queue' —— 1.4.4 的真实回归），反过来项目自己的 模块也可能被同名的已安装包顶掉。注册包对象只影响 app.* 这一条解析链，进程里其他名字都不受 项目内容影响。

代价：扁平导入（import model.user）不再可用（脚手架与本文档一直用的是 app. 形式）。 若你的项目依赖扁平导入，在自己的 bootstrap/main.py 里补一行 sys.path.append("app") 即可 —— 但请知道那会把该目录下的名字暴露给整个进程。

配置

所有配置在 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()

# 控制器方法引用（推荐，IDE 可跳转）
router.get("/users", UserController.index)
router.get("/users/{id}", UserController.show)


# 普通函数 / lambda / 闭包
async def health(request): ...


router.get("/health", health)

# 路由描述（写入 OpenAPI summary，/docs 可见）
router.get("/index", HomeController.index, intro="首页")

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

# 单条路由挂中间件（最内层，最贴近端点；顺序：Router 级 → 分组级 → 路由级）
router.get("/orders", OrderController.index, middleware=[auth()])

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

中间件签名是 async def __call__(self, request, call_next)，拿到的是框架 Request （request.bearer_token() / header() / json() / state 都可用，也可直接 request.raw 取底层 Starlette 请求）；不调用 call_next 即可短路返回响应。

路径参数、查询参数、请求体经 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()。

请求信息：request.method / request.path / request.url（字符串）/ request.headers（大小写不敏感） 按属性访问（与 Starlette 一致），旧的方法写法 request.method() 仍然可用。 request.url 是字符串（Starlette 那边是 URL 对象），取路径请用 request.path。

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


# 继承顺序必须把 Mixin 放在 Model 前面：反过来 MRO 会把 delete() 解析到
# Model.delete（物理删除），软删除静默失效。make:model 生成的骨架就是这个顺序。
class User(SoftDeletes, TimestampsMixin, Model):
    __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)
    # 停用字段：auth 守卫登录/校验 token 时读它（字段名可在 config/auth.py 的
    # providers.users.active_field 改）。模型上没有该字段时守卫一律放行，
    # 表现为"停用不了用户"，启动日志会给出 warning。
    is_active: Mapped[int] = f.boolean(default=True)

    # 访问器：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()  # 永久删除

# 批量删除同样遵循软删除（与 Laravel 一致）；物理删除用 force_delete()
await User.query().where("status", 0).delete()  # 软删除命中行
await User.query().where("status", 0).force_delete()  # 物理删除命中行
await User.destroy(1, 2, 3)  # 软删除模型 → 软删除

# 关联（含 with_ 急加载、where_has）默认也排除已软删的行
posts = await user.posts().get()  # 不含已软删
posts = await user.posts().with_trashed().get()  # 含已软删

关联

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, auth, auth_optional

# 登录
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)
get_auth().user()  # 当前用户，按请求/task 隔离

# 登出 + 吊销（吊销后该 token 立即失效，不必等过期）
guard.logout()
result = await guard.revoke(token)
if not result.shared:
    # 未落到共享存储：未配 Redis（单进程部署下即为全局）或 Redis 写失败（降级）
    # degraded=True 时多实例部署下其他实例仍认这个 token，应告警或返回 503
    Log.warning(f"token 吊销未全局生效: {result.reason}")

# 密码哈希（自动升级旧算法，BCRYPT_ROUNDS 默认 12；上限 72 字节）
from fastv.auth import PasswordHasher

hasher = PasswordHasher()
hasher.make("plain")
hasher.check("plain", hashed)

保护路由（解析 Authorization: Bearer → 校验 → 设为当前用户）：

# app/route/web.py
from fastv.auth import auth, auth_optional
from fastv.routing.router import Router

router = Router()

with router.group(prefix="/admin", middleware=[auth()]) as g:
    g.get("/dashboard", AdminController.dashboard)

router.get("/orders", OrderController.index, middleware=[auth()])  # 或按路由挂

router.get("/feed", FeedController.index, middleware=[auth_optional()])  # 可选登录

用户模型契约：守卫用 find_by_credentials(credentials) 按凭证查人。 继承 fastv.database.model.Model 即有默认实现（取凭证里除 password 外的字段做 AND 查询，软删除用户自动排除），需要多字段登录时覆盖它即可：

class User(Model):
    @classmethod
    async def find_by_credentials(cls, credentials):
        return await cls.where("phone", credentials["phone"]).first()

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

Token 吊销名单优先写 Redis（多进程共享，TTL = token 剩余有效期）；未配 Redis 时退化为进程内记录（仅单进程有效）。Redis 故障期间吊销检查是 fail-open —— 与框架其它 Redis 依赖一致，不会因 Redis 抖动把所有登录一并否决。

缓存

三驱动（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（不拖垮业务）。

每个 store 都要配 prefix —— cache:clear 的删除边界就是它：redis store 没有 prefix 时不存在安全边界（scan "*" 会把同一个 Redis 库里的队列键、分布式锁、限流计数一并删掉），命令会直接拒绝执行并提示补配；file store 的 path 是相对路径时按项目根目录（而非进程 CWD）解析，flush 也只删本 store 前缀下的文件。

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"

    def __init__(self, to: str):
        super().__init__(to=to)

    async def handle(self):
        to = self.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
fastv queue:stats                     # 各队列深度与计数

失败任务：

任务重试次数耗尽后会写入数据库表 fail_jobs（参考 Laravel 的 failed_jobs； 载荷 base64 存放，重试时原样回队），因此与用哪个驱动无关 —— redis 驱动 本身不保留失败任务体，靠这张表补齐查看与重试能力：

fastv queue:failed-table    # 生成 fail_jobs 建表迁移（含 created_at 索引）
fastv migrate               # 建表
fastv queue:failed          # 列出失败任务（任务 ID / 任务类 / 队列 / 失败时间 / 异常）
fastv queue:retry <task_id|行 id>   # 重新入队，成功后删除该行
fastv queue:retry           # 不带参数 = 重试全部（最多 1000 条）
fastv queue:flush           # 清空失败任务表（不带 --force 会先确认）

表已存在时 queue:failed-table 只提示、不生成迁移；若该表缺 created_at 索引，会直接打印可执行的 CREATE INDEX 语句。

崩溃遗留任务回收（仅 redis 驱动）：

任务被 dequeue 后会搬到 queue:{前缀}{队列}:processing。worker 被 kill / 重启时 （部署、OOM、容器重启）它会永久留在那里 —— 内存只增不减、queue:stats 的 Running 永久虚高。fastv 在 dequeue 时给每条任务记一个到期时间，worker 启动时与 后台循环里把超时的搬回主队列：

fastv queue:recover                 # 回收超时未归的任务（全部队列）
fastv queue:recover emails          # 只回收某个队列
fastv queue:recover --force         # 不看时间戳整表搬回（确认没有存活 worker 时用）
fastv queue:recover --timeout=60    # 只捞卡了 60 秒以上的

超时阈值与扫描间隔在 config/queue.py 的 asynctasq.redis 段配置 （processing_timeout 默认 3600s，应设为最长任务耗时的 2~3 倍；reclaim_interval 默认 30s，设 0 则只在启动时回收一次）。回收是 at-least-once：跑得比 processing_timeout 更久的任务可能被重复执行一次（与 SQS/MySQL/Postgres 驱动的 visibility timeout 同语义）。queue:stats 里的 Stale 就是等待回收的条数。

表未创建或数据库不可用时，记录降级为"打一条 warning、不落库"，不影响任务消费。

worker 心跳：

queue:stats 的 Workers 段落与 Queue.workers() 数据来自 worker 心跳 （worker:{前缀}{id} 哈希，写库人是 Worker 自己，键带 TTL，异常退出后自动过期）； key_prefix 不同则互不可见，多项目共用同一个 Redis 也不会串。

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

    # ---- 限速配置（对齐上游 API 上限）----
    rate_limit = 10  # 窗口内最多执行 10 次
    rate_limit_window = 60  # 窗口 60 秒 → 10 QPM（每分钟 10 张）
    rate_limit_timeout = 300  # 排队等配额最多 5 分钟，超时抛异常走重试
    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()

限速参数换算：

重试间隔由全局配置决定：任务上的 retry_delay_seconds / .retry_after() 不生效 （dispatch 时会 WARNING 提示一次）。asynctasq worker 只用 queue.asynctasq.task_defaults.retry_delay 配合 retry_strategy： fixed = 每次固定间隔，exponential = retry_delay × 2^(已重试次数-1)。

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

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)  # 触发重试（间隔见全局 task_defaults.retry_delay）

生图等上游限流场景推荐 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 锁（带 token 的 SET NX EX，释放走 Lua CAS，不会删掉别人的锁）， 多 worker 部署时保证同一定时任务只执行一次：Redis 不可用时该任务会被跳过（宁可晚跑也不重复跑， /health 之外会打 error 日志）；without_overlapping() 任务运行期间跳过重叠触发，Redis 不可用时 降级为进程内锁并明确告警（此时多进程仍可能同时执行）。

事件

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

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,
    // 路径必须与服务端 config/websocket.py 的 path 一致：
    // pusher-js 默认连 /app/{key}，而服务端默认是 /ws，不写 wsPath 会连不上
    wsPath: '/ws',
    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 桥）；桥断线会自动退避重连，重连窗口内本进程的广播回退为进程内直投（本进程客户端不丢消息；窗口内其他实例收不到，这是 Pub/Sub 的固有限制）。

presence 通道（presence-*）会跟随连接生命周期维护在线成员：同一用户多连接只在首个连接加入时广播 member_added、最后一个连接断开时才广播 member_removed。

中间件

框架内置安全中间件（参数 .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_*，默认关闭）。CORS_ALLOW_CREDENTIALS=true 与 CORS_ORIGINS=* 同时启用时生产环境拒绝启动（任意站点可携带用户 Cookie 跨域读接口）
• ThrottleMiddleware — 限速（Redis 滑动窗口，TRUSTED_PROXIES 白名单防伪造 IP）
  • 限流键按 IP 分桶（等价 Laravel 挂在路由组上的语义），不含路径 —— 按路径分桶会让 客户端变换 URL 就拿到无限配额；/health、/health/ready、/storage 默认不参与限流 （探针与静态资源不该抢业务配额），可用 THROTTLE_SKIP_PATHS 覆盖
  • SECURITY_HSTS 的"仅在 HTTPS 上下发"走可信代理推导（TRUSTED_PROXIES 内的 X-Forwarded-Proto）：反代终止 TLS 时后端看到的是 http，不这样处理 HSTS 永远不会下发

限速（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（默认关闭）

发布（维护者）

发布走脚本，避免"改了版本号没提交 / 构建产物是上一次的 / 漏打 tag"这类发布事故：

# 一次性准备：把 PyPI API token 填进 .env 的 PYPI_TOKEN
# （PyPI → Account settings → API tokens，scope 选 project: fastv）
# .env 已被 .gitignore 忽略；脚本还会拒绝从"被 git 跟踪的 .env"读凭证

python scripts/release.py 1.4.5                    # 升版本 → lint/测试 → 提交+tag → 构建 → 校验 → 上传
python scripts/release.py 1.4.5 --no-upload        # 演练：到校验为止，不发布
python scripts/release.py 1.4.5 --notes notes.md   # 把发布说明作为提交信息正文
python scripts/release.py --reupload               # 产物已构建，只重试上传
python scripts/release.py 1.4.5 --push             # 上传成功后自动推 master 与 tag

脚本按固定顺序执行，每步失败都回滚到"可以重来"的状态：要求工作区干净、当前分支为 master、目标版本大于当前版本；上传用的 token 经环境变量传给 twine（不落命令行， 避免被进程列表看到）。上传不可撤销（同一版本号不能重传），所以它先本地提交并打 tag， 上传成功后才提示（或执行）推送。

许可证

MIT