Metadata-Version: 2.4
Name: fastv
Version: 1.5.5
Summary: Laravel-style FastAPI framework
Author-email: FastV Team <support@fastv.dev>
License-Expression: MIT
Keywords: fastapi,laravel,web-framework,async,orm,cli
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.12
Description-Content-Type: text/markdown
Requires-Dist: fastapi<0.200,>=0.139.0
Requires-Dist: uvicorn<1.0,>=0.51.0
Requires-Dist: pydantic<3.0,>=2.13.0
Requires-Dist: sqlalchemy<2.1,>=2.0.0
Requires-Dist: python-dotenv<2.0,>=1.2.0
Requires-Dist: click<9.0,>=8.4.0
Requires-Dist: rich<16.0,>=14.0.0
Requires-Dist: redis<9.0,>=5.0
Requires-Dist: apscheduler<4.0,>=3.10.0
Requires-Dist: alembic<2.0,>=1.13.0
Requires-Dist: PyJWT<3.0,>=2.8.0
Requires-Dist: passlib[bcrypt]<2.0,>=1.7.4
Requires-Dist: bcrypt<4.1.0,>=4.0.0
Requires-Dist: asynctasq<2.0,>=1.7.0; sys_platform != "win32"
Requires-Dist: python-multipart<1.0,>=0.0.7
Requires-Dist: psutil<8.0,>=5.9.0
Requires-Dist: watchfiles<2.0,>=1.0.0
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.24; extra == "dev"
Requires-Dist: pytest-cov>=4.0; extra == "dev"
Requires-Dist: httpx>=0.28.0; extra == "dev"
Requires-Dist: ruff>=0.8.0; extra == "dev"
Requires-Dist: fakeredis[lua]>=2.20; extra == "dev"
Provides-Extra: mysql
Requires-Dist: asyncmy>=0.2.9; extra == "mysql"
Requires-Dist: pymysql>=1.0; extra == "mysql"
Requires-Dist: cryptography>=3.0; extra == "mysql"
Provides-Extra: postgres
Requires-Dist: asyncpg>=0.30.0; extra == "postgres"
Requires-Dist: psycopg2-binary>=2.9; extra == "postgres"
Provides-Extra: sqlite
Requires-Dist: aiosqlite>=0.20.0; extra == "sqlite"
Provides-Extra: oss
Requires-Dist: alibabacloud-oss-v2>=1.3.0; extra == "oss"
Provides-Extra: permission
Requires-Dist: pycasbin>=2.0; extra == "permission"
Requires-Dist: casbin-async-sqlalchemy-adapter>=1.9; extra == "permission"
Provides-Extra: all
Requires-Dist: asyncmy>=0.2.9; extra == "all"
Requires-Dist: pymysql>=1.0; extra == "all"
Requires-Dist: cryptography>=3.0; extra == "all"
Requires-Dist: asyncpg>=0.30.0; extra == "all"
Requires-Dist: psycopg2-binary>=2.9; extra == "all"
Requires-Dist: aiosqlite>=0.20.0; extra == "all"
Requires-Dist: alibabacloud-oss-v2>=1.3.0; extra == "all"
Requires-Dist: pycasbin>=2.0; extra == "all"
Requires-Dist: casbin-async-sqlalchemy-adapter>=1.9; extra == "all"

# FastV — Laravel-style FastAPI Framework

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

- **快速上手**：[安装](#安装) → [快速开始](#快速开始) → [目录结构](#目录结构)
- **核心开发**：[配置](#配置) → [路由](#路由) → [控制器](#控制器) → [模型 ORM](#模型) → [验证](#验证) → [认证](#认证)
- **基础设施**：[缓存](#缓存) → [Redis](#redis) → [文件存储](#文件存储) → [日志](#日志)
- **异步能力**：[队列](#队列) → [调度](#调度) → [事件](#事件) → [WebSocket 广播](#websocket-广播)
- **运维**：[中间件](#中间件) → [异常处理](#异常处理) → [CLI 命令](#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 + 连接池水位）

---

## 安装

```bash
# 核心框架
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+。

## 快速开始

```bash
# 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")`。

```python
# 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": ...}  云数据库强制加密
    },
}
```

代码中读取配置：

```python
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`，自动发现并注册。三种写法：

```python
# 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 原生注入：

```python
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`，自动注入当前请求，提供响应/异常快捷方法。

```python
# 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` 生成。

```python
# 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）

```python
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`（悲观锁）。

```python
# 分页
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()
```

### 写入

```python
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)
```

### 软删除

```python
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()  # 含已软删
```

### 关联

```python
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()
```

### 死锁重试

```python
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 风格字符串规则与自定义规则。

```python
# 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 保护）
```

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

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

**自定义规则**：

```python
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 密码哈希。

```python
# 注册（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` → 校验 → 设为当前用户）：

```python
# 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 查询，软删除用户自动排除），需要多字段登录时覆盖它即可：

```python
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 降级）。

```python
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），生产级：连接验证、故障快速失败。

```python
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）双驱动。

```python
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`：
```python
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 结构化可选。

```python
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 的任务队列。

```python
# 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：
```bash
fastv queue:work --queue=default,emails --concurrency=10
fastv queue:stats                     # 各队列深度与计数
```

失败任务：

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

```bash
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 启动时与
后台循环里把超时的搬回主队列：

```bash
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 固定窗口计数器），窗口释放后自动继续，不烧重试次数。

```python
# 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)`。

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

```python
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`）：
```python
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。

```python
# 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")` 等。

启动守护进程：
```bash
fastv schedule:serve
```

`on_one_server()` 基于 Redis 锁（带 token 的 `SET NX EX`，释放走 Lua CAS，不会删掉别人的锁），
多 worker 部署时保证同一定时任务只执行一次：**Redis 不可用时该任务会被跳过**（宁可晚跑也不重复跑，
`/health` 之外会打 error 日志）；`without_overlapping()` 任务运行期间跳过重叠触发，Redis 不可用时
降级为进程内锁并明确告警（此时多进程仍可能同时执行）。

---

## 事件

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

```python
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`。

### 后端广播

```python
# 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})
```

### 通道授权

```python
# 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）

```js
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）+ 自定义中间件自动发现。

```python
# 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 注册）：
```python
global_middleware = [
    "fastv.http.middleware.throttle.ThrottleMiddleware",
]
```

---

## 异常处理

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

```python
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 命令

```bash
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` 自动发现）：
```python
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"这类发布事故：

```bash
# 一次性准备：把 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
