Skip to main content

FastV — Laravel-style FastAPI Framework

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


特性

  • 服务容器(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="首页")

# 分组(/docs 里的 tag):三级都可用,合并顺序 Router 级 → 分组级 → 路由级
#   分组顺序与描述由 config/app.py 的 openapi_tags 决定(列表顺序即分组顺序)
router = Router(prefix="/admin", tags=["后台管理"])  # 整个文件
router.get("/users", UserController.index, tags=["管理员"])  # 单条路由
router.group(prefix="/asset", tags=["素材库"], callback=lambda g: [...])

# 分组(闭包 / 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

这些注解同时决定 /docs 里的参数文档:

方法签名 /docs 里的呈现
async def show(self, id: int) path 参数(标量注解由 FastAPI 原生生成)
async def index(self, page: int = 1) query 参数(含类型与默认值)
async def store(self, data: UserForm) UserForm 的 rules() 展开成 requestBody(GET 时展开成 query 参数)

FormRequest 注解是框架自己解析的(合并 query/form/JSON 再校验),FastAPI 看不见 它,所以由框架把它按 rules() 展开进 /docs:字段、必填、约束(max:20 → maxLength)、 messages() 的中文提示(→ 字段 description)都会出现,且与运行时校验同源、不会漂移。

方法体内运行时取值(await self.request.integer("page", 1))无法被文档记录 —— 没有声明就无从推断,需要出现在 /docs 里就得写进方法签名。

/docs 的标题、版本、描述、分组顺序、OpenAPI 规范版本取自 config/app.py 的 title / version / description / openapi_tags / openapi_version(都有对应环境 变量)。title 未配时回退到 name(所以标题默认显示项目名),都没有才是 FastAPI 的 "FastAPI Application";版本与描述不配就是 "0.1.0" 与空;分组不配按字母序; openapi_version 默认 "3.0.3"(3.1 的 JSON Schema 2020-12 语义,部分代码生成器 / 旧版工具还不认),消费方都支持时改成 "3.1.0"。

规范版本不只改标签:Pydantic v2 生成的是 JSON Schema 2020-12,FastAPI 自己不会为 3.0 转换,所以框架会在 3.0.x 下把整份文档降级成严格符合 3.0 的写法 —— 可选字段从 anyOf: [X, {"type": "null"}] 变成 nullable: true(X 是 $ref 时包一层 allOf), const → enum、contentMediaType → format: binary、数值型 exclusiveMinimum/Maximum → 布尔型 + minimum/maximum、examples → example、定长元组退化成"元素是这些类型之一" (3.0 的 items 必须是对象、不能是数组,元组的位置信息在 3.0 里表达不了),并去掉 3.1 专有的顶层 webhooks。 同时把内联 schema 里的 $defs 提升到 components/schemas(#/$defs/… 这种相对文档根 的指针内联后会悬空,两种版本下都解析不出来)。用 openapi_version = "3.1.0" 则保持 FastAPI 原生输出、不做任何改写。

文档里还会自动补上这些语义项(空占位不写 —— OpenAPI 允许省略):

项 来源
security: [{"bearer": []}] + components.securitySchemes.bearer 挂了 auth() 中间件的接口(auth_optional() 与公开接口不写)
servers config('app.url')(APP_URL),不配则不写
operation 的 summary / description intro= / 控制器方法的 docstring(intro 也可省,FastAPI 会由函数名生成)
requestBody 的 content 类型 带 FormRequest 表单的接口统一声明 application/x-www-form-urlencoded;控制器形参写了 UploadFile(有文件上传)时用 multipart/form-data —— 文件只能走 multipart,且一个请求只能有一种表单编码

控制器

控制器继承 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


# 继承顺序不影响语义:delete() 由 Model 按 DELETED_AT 标记判定软删还是物理删,
# 所以 (SoftDeletes, TimestampsMixin, Model) 与 (Model, SoftDeletes) 行为一致。
# 这里把 Mixin 写前面只是读起来更直观(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()

限速参数换算:

上游 API 限制 配置
每分钟 N 张 rate_limit=N, rate_limit_window=60
每秒 N 张 rate_limit=N, rate_limit_window=1
每 X 秒 1 张 rate_limit=1, rate_limit_window=X
高峰积压 rate_limit_timeout 调大或 None(无限排队,不失败)

重试间隔由全局配置决定:任务上的 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

Metadata

Release files for fastv 1.5.5

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for fastv 1.5.5
File Size Uploaded
fastv-1.5.5.tar.gz 363.7 kB Details

Built distribution (wheel)

Table of built distributions (wheels) for fastv 1.5.5
File Interpreter ABI Platform
fastv-1.5.5-py3-none-any.whl Python 3 none any Details

Total release size: 761.0 kB

Release files / fastv-1.5.5.tar.gz

Download URL fastv-1.5.5.tar.gz
Size 363.7 kB
Tags Source
SHA-256 checksum
How to use checksums
91252b291f2e43fab7570694d80dc8b7746cdbc24edc1c614510aa37e749faf6
BLAKE2b-256 checksum
How to use checksums
640a7ea2e87186995e37aa24272cbfea5a94cf46d46eabc8213e223fcc4c20eb
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via twine/7.0.0 CPython/3.13.14

Release files / fastv-1.5.5-py3-none-any.whl

Download URL fastv-1.5.5-py3-none-any.whl
Size 397.4 kB
Tags Python 3
SHA-256 checksum
How to use checksums
3629ffd9b5a44b6fdd5176c28b175aaf0e16eeaf420924a14dd71745ee6b9eea
BLAKE2b-256 checksum
How to use checksums
ed5df76003774c09e2bdd4d39211438d06c96e995540ea999bf863ec220cb5bf
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via twine/7.0.0 CPython/3.13.14

Release history Release notifications | RSS feed

1.6.0

2 release files

1.5.9

2 release files

1.5.8

2 release files

1.5.7

2 release files

1.5.6

2 release files

This release

1.5.5 This release

2 release files

1.5.4

2 release files

1.5.3

2 release files

1.5.2

2 release files

1.5.1

2 release files

1.5.0

2 release files

1.4.9

2 release files

1.4.8

2 release files

1.4.7

2 release files

1.4.6

2 release files

1.4.5

2 release files

1.4.4

2 release files

1.4.2

2 release files

1.4.1

2 release files

1.4.0

2 release files

1.3.1

2 release files

1.3.0

2 release files

1.2.9

2 release files

1.2.8

2 release files

1.2.7

2 release files

1.2.6

2 release files

1.2.5

2 release files

1.2.4

2 release files

1.2.3

2 release files

1.2.2

2 release files

1.2.1

2 release files

1.2.0

2 release files

1.1.9

2 release files

1.1.8

2 release files

1.1.7

2 release files

1.1.6

2 release files

1.1.5

2 release files

1.1.4

2 release files

1.1.3

2 release files

1.1.2

2 release files

1.1.1

2 release files

1.1.0

2 release files

1.0.4

2 release files

1.0.3

2 release files

1.0.1

2 release files

1.0.0

2 release files

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page