Metadata-Version: 2.4
Name: fastv
Version: 1.1.8
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.139.0
Requires-Dist: uvicorn>=0.51.0
Requires-Dist: pydantic>=2.13.0
Requires-Dist: sqlalchemy>=2.0.0
Requires-Dist: python-dotenv>=1.2.0
Requires-Dist: click>=8.4.0
Requires-Dist: rich>=14.0.0
Requires-Dist: redis>=5.0
Requires-Dist: apscheduler>=3.10.0
Requires-Dist: alembic>=1.13.0
Requires-Dist: PyJWT>=2.8.0
Requires-Dist: passlib[bcrypt]>=1.7.4
Requires-Dist: bcrypt<4.1.0,>=4.0.0
Requires-Dist: asynctasq>=1.7.0; sys_platform != "win32"
Requires-Dist: python-multipart>=0.0.7
Requires-Dist: psutil>=5.9.0
Requires-Dist: watchfiles>=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"
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"

# 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）。

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

---

## 配置

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

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

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

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

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

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

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

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

---

## 控制器

控制器继承 `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()`。

**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

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

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

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

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

### 查询（链式 QueryBuilder）

```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()                         # 永久删除
```

### 关联

```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, AuthManager

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

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

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

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

---

## 缓存

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

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

---

## 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"

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

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

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

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

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

启动 Worker：
```bash
fastv queue:work --queue=default,emails --concurrency=10
```

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

---

## 调度

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 锁（`SET NX EX`），多 worker 部署时保证同一定时任务只执行一次；`without_overlapping()` 任务运行期间跳过重叠触发。

---

## 事件

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

```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,
    forceTLS: false, disableStats: true,
    auth: { headers: { Authorization: 'Bearer <jwt>' } },
});

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

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

---

## 中间件

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

```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_*`，默认关闭）
- `ThrottleMiddleware` — 限速（Redis 滑动窗口，`TRUSTED_PROXIES` 白名单防伪造 IP）

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

## 许可证

MIT
