Metadata-Version: 2.4
Name: socar-api
Version: 0.7.13
Summary: 基于 FastAPI + JSON:API 规范的 RESTful 接口框架
Author-email: jidongbo <jidongbo@gmail.com>
License: MIT
Requires-Python: >=3.12
Description-Content-Type: text/markdown
Requires-Dist: fastapi>=0.136.3
Requires-Dist: pydantic>=2.13.4
Requires-Dist: jquery-unparam>=2.0.0
Requires-Dist: pyyaml>=6.0.3
Requires-Dist: treelib>=1.8.0
Requires-Dist: bitarray>=3.8.1
Requires-Dist: python-multipart>=0.0.32
Requires-Dist: python-dateutil>=2.9.0.post0
Requires-Dist: typing-extensions>=4.15.0
Requires-Dist: portalocker>=3.1.0
Provides-Extra: dev
Requires-Dist: pytest>=9.0.3; extra == "dev"
Requires-Dist: pytest-asyncio>=1.4.0; extra == "dev"
Requires-Dist: httpx>=0.28.1; extra == "dev"
Requires-Dist: pytest-cov>=7.1.0; extra == "dev"
Requires-Dist: ruff>=0.15.16; extra == "dev"
Requires-Dist: mypy>=2.1.0; extra == "dev"
Requires-Dist: pre-commit>=4.6.0; extra == "dev"
Provides-Extra: data
Requires-Dist: sqlalchemy>=2.0; extra == "data"
Requires-Dist: redis>=5.0; extra == "data"
Provides-Extra: clickhouse
Requires-Dist: clickhouse-sqlalchemy>=0.3.0; extra == "clickhouse"
Requires-Dist: clickhouse-driver>=0.2.0; extra == "clickhouse"

# socar_api

基于 FastAPI + JSON:API 规范的 RESTful 接口框架。

[![version](https://img.shields.io/badge/version-0.7.13-blue.svg)](https://pypi.org/project/socar-api/0.7.13/)

## 简介

socar_api 提供了从资源声明 → 路由生成 → 请求处理 → 响应序列化的完整链路，让开发者**专注业务逻辑**，无需重复实现接口规范。

框架提供两种资源基类：
- **`BaseResource`** — 完整基类，支持 `connect_data()` session 管理和子资源嵌套
- **`Resource`**（推荐） — 简化基类，内置 JSON:API 序列化管线，可直接访问 `self.db`

```python
from socar_api import Resource, SchemaBase, Field, Relationship

class ArticleModel(SchemaBase):
    title: str = Field(None)
    content: str = Field(None)
    author_id: int = Field(None, isrel=True)

class Article(Resource):
    model = ArticleModel

    class Meta:
        link = '/articles'
        type_ = 'articles'

    class RelResources:
        author = Relationship(
            rel_resource='Author',
            mapping_field='author_id',
            has_api=False,
        )

    async def get_many(self):
        return self.db.query(ArticleModel).all()

    async def post(self):
        body = self.parse_body()
        obj = ArticleModel(**body.model_dump())
        self.db.add(obj)
        self.db.commit()
        return self.model.model_validate(obj)
```

一行 `Article.register_routes(app)` 即生成完整 JSON:API 接口。

## 核心特性

- **JSON:API 规范** — 请求/响应格式、资源关系、稀疏字段、include 复合文档、分页
- **资源路由** — 类声明式路由，自动生成标准 CRUD 端点
- **关系管理** — 一对一/一对多关系，include 深层级联（最多支持 3 层）
- **权限系统** — 基于 scope 的 OAuth2 认证，API + 关系双层权限；支持 `@scope` 声明式配置
- **原子操作** — JSON:API Atomic 扩展支持，批量操作 + 事务回滚
- **GET 批量聚合** — `POST /_batch` 单请求执行多个 GET 子请求，并发/串行双模式，独立 DB session 隔离
- **文件上传/下载** — 增强的 `UploadFileResource` / `DownloadFileResource` 基类，文件类型验证、关系校验、可替换存储后端
- **响应缓存** — `@cached()` 装饰器一行启用接口缓存，Cache-Aside 模式，可替换后端
- **筛选过滤** — Deep Object 格式，40+ 操作符覆盖字符串/数值/列表/时间/布尔；filter 模型自动暴露到 OpenAPI schema；支持 `required_filters` 必填标注和 `description` 中文说明（v0.7.10）
- **自动模型** — 根据资源自动生成请求/响应模型
- **资源版本** — 内建 API 版本管理
- **自动文档** — 与 FastAPI OpenAPI 集成

## 安装

```bash
pip install socar-api
```

> **最低要求**: Python 3.12+

## 第一个资源

### 1. 定义数据模型

```python
from socar_api import SchemaBase, Field

class BookModel(SchemaBase):
    id: str = Field(None)
    title: str = Field(None)
    author: str = Field(None, title='作者')
    price: float = Field(None)
```

> **类型转换**: `SchemaBase` 自动将 `int` 值转为 `str` 字段（如 `id=1` → `"1"`），兼容 Pydantic v1 行为。
> 如需禁用此行为以获得严格类型检查，使用 `StrictSchemaBase`：
>
> ```python
> from socar_api import StrictSchemaBase, Field
>
> class BookModel(StrictSchemaBase):
>     id: str = Field(None)  # id=1 将抛出 ValidationError
> ```

### 2. 定义资源

```python
from socar_api import BaseResource

class Book(BaseResource):
    model = BookModel
    methods = {'GET', 'GETS', 'POST', 'PATCH', 'DELETE'}

    class Meta:
        link = '/books'
        type_ = 'books'
```

### 3. 创建应用

```python
from fastapi import FastAPI
from socar_api.util import register_jsonapi_exception_handlers, JSONAPIContentNegotiationMiddleware
from socar_api import BaseResource

app = FastAPI()
register_jsonapi_exception_handlers(app)           # JSON:API 格式错误响应
app.add_middleware(JSONAPIContentNegotiationMiddleware)  # Content-Type 校验（可选）

Book.register_routes(app)
```

### 4. 启动

```bash
uvicorn main:app --reload
```

访问 `http://localhost:8000/docs` 查看自动生成的接口文档。

启动后自动生成：

| 方法 | 路径 | 说明 |
|------|------|------|
| `GET` | `/books` | 资源列表 |
| `GET` | `/books/{id}` | 单个资源（需实现 `get()` 方法） |
| `POST` | `/books` | 新增资源 |
| `PATCH` | `/books/{id}` | 更新资源 |
| `DELETE` | `/books/{id}` | 删除资源 |
| `POST` | `/atomic` | 原子操作（`register_routes()` 或 `setup_app(with_atomic_endpoint=True)` 自动注册） |

> **注意:** 框架会自动检测子类是否真实覆写了 handler 方法（`get_many`/`get`/`post`/`patch`/`delete` 及关系 `rel_*` 方法）。声明了对应 `methods` 但未实现的死路由会被跳过并输出 warning。因此声明 `methods` 后务必实现对应的 handler 方法。

## 完整的 CRUD 资源

### 实现业务逻辑

```python
from socar_api import BaseResource, SchemaBase, Field, Relationship


class ArticleModel(SchemaBase):
    id: str = Field(None)
    title: str = Field(None)
    content: str = Field(None)
    status: str = Field(None)
    author_id: str = Field(None, isrel=True)


class Article(BaseResource):
    model = ArticleModel
    methods = {'GET', 'GETS', 'POST', 'PATCH', 'DELETE'}

    class Meta:
        link = '/articles'
        type_ = 'articles'

    class RelResources:
        author = Relationship(
            rel_resource='Author',
            mapping_field='author_id',
            has_api=False,
        )

    async def get_many(self):
        skip = self.args.skip
        limit = self.args.limit
        query = self.db.query(ArticleModel)
        return query.offset(skip).limit(limit).all()

    async def get(self):
        id_ = self.request.path_params.get('id')
        return self.db.query(ArticleModel).filter_by(id=id_).one()

    async def post(self):
        body = self.parse_body()
        obj = ArticleModel(**body.model_dump())
        self.db.add(obj)
        self.db.commit()
        return obj

    async def patch(self):
        body = self.parse_body()
        obj = self.db.query(ArticleModel).filter_by(id=body.id).one()
        for key, value in body.model_dump(exclude_unset=True).items():
            setattr(obj, key, value)
        self.db.commit()
        return obj

    async def delete(self):
        body = self.parse_body()
        obj = self.db.query(ArticleModel).filter_by(id=body.id).one()
        self.db.delete(obj)
        self.db.commit()
        return obj
```

> `self.db` 由 `BaseResource` 自动注入，CRUD 方法中可直接使用，无需额外包裹。

## 关系资源

```python
from socar_api import BaseResource, SchemaBase, Field, Relationship


class AuthorModel(SchemaBase):
    id: str = Field(None)
    name: str = Field(None)


class Author(BaseResource):
    model = AuthorModel

    class Meta:
        link = '/authors'
        type_ = 'authors'

    class RelResources:
        articles = Relationship(
            rel_resource='Article',
            cond_fun='author_articles',
            one_to_one=False,
            has_api=False,
        )
```

自动生成的关系接口：

- `GET /authors/{id}/relationships/articles` — 获取关系
- `GET /authors/{id}/articles` — 获取相关资源
- `GET /articles?include=author` — 在文章列表中关联作者

### 参数型关系

当关系数据来自请求参数（如分析型资源通过 `filter[group]` 计算）而非数据库字段时，使用 `source='param'`，框架将：

- 不挂载 `/authors/{id}/relationships/articles` 与 `/authors/{id}/articles` 路由
- 不生成 `self`/`related` links
- 禁止通过 JSON:API 关系负载写入（`modify=False`）
- 仍保留 schema 模型，支持 `include`

```python
class Sales(BaseResource):
    class RelResources:
        group = Relationship(
            rel_resource='Options',
            mapping_field='group',
            source='param',
        )
```

## 查询与筛选

框架自动解析 URL 参数，支持 Deep Object 格式：

```
GET /articles?filter[title][op]=ct&filter[title][value]=Python
GET /articles?sort=-created_at
GET /articles?page[offset]=0&page[limit]=20
GET /articles?include=author.comments
GET /articles?fields[articles]=title,content
```

等价于编程式筛选：

```python
from socar_api.data.spec import Filter, FilterAnd, FilterOr

filter = FilterAnd([
    Filter(field='title', op='ct', value='Python'),
    Filter(field='status', op='eq', value='published'),
])
```

> **v0.6.3+**：每个 filter 字段的合法操作符集合和值类型已暴露到 OpenAPI schema 中。v0.7.12+ 同时为每个 filter 参数注入 `example` 值。前端 codegen 可以直接从 `paths/{resource}/get/parameters` 读取 `filter[field]` 参数定义。

## 数据库与 Repository

安装数据库支持：

```bash
pip install socar-api[data]      # SQLAlchemy + Redis
pip install socar-api[clickhouse] # ClickHouse 支持
```

### 初始化

```python
from socar_api import setup_app

# YAML 配置示例（config.yaml）
# database:
#   mysql:
#     host: localhost
#     port: 3306
#     user: root
#     password: xxx
#     database: mydb
#     charset: utf8mb4
#   mysql_platform:
#     host: 10.0.0.1
#     user: admin
#     password: xxx
#     database: platform
#   clickhouse:
#     host: ch.example.com
#     port: 9000
#     user: default
#     password: xxx
#     database: analytics

from fastapi import FastAPI
from socar_api.config import Config

app = FastAPI()

config = Config()
config.load('config.yaml')

# 一键初始化：DB 连接 + 中间件 + CORS + 异常处理 + batch/atomic 端点
# 注意：必须在所有 register_routes() 调用之后执行 setup_app()
setup_app(app, config)
```

`init_db` 自动扫描 YAML 中所有 `database.mysql.*` 和 `database.clickhouse.*` 配置节，注册为命名 session（`db_mysql`、`db_mysql_platform`、`db_clickhouse` 等），`db_middleware` 按请求自动创建/收尾/关闭所有已注册的 session。如需接入 MongoDB、Redis 等非 SQLAlchemy 数据库，见下方"数据库扩展机制"。

### 多库与命名 Session

资源可通过 `session_name` 指定使用的数据库实例：

```python
from socar_api import BaseResource, Field, SchemaBase

class PlatformModel(SchemaBase):
    id: str = Field(None)
    name: str = Field(None)


class Platform(BaseResource):
    model = PlatformModel
    session_name = 'db_mysql_platform'  # 指定使用 platform 库

    class Meta:
        link = '/platforms'
        type_ = 'platforms'
```

`connect_data()` 查找优先级：`session_name` > `db_{session_type}` > `None`。

### Repository 模式

使用 `repo()` 工厂一行创建数据库操作层，替代手写 CRUD 文件：

```python
from socar_api.data.repository import repo

repo = repo(MyOrmModel, MySchema)
# 等价于：
# class CrudMyModel(DataCrud):
#     orm = MyOrmModel
#     schema = MySchema

repo.get_one(db=session, id_=1)
repo.get_multi(db=session, filters=filter, skip=0, limit=100, sort=sort)
repo.create_one(db=session, data=schema_instance)
repo.update_one(db=session, id_=1, data=schema_instance)
repo.count(db=session, filters=filter)
repo.delete_one(db=session, id_=1)
```

在资源类中使用：

```python
from socar_api.data.repository import repo
from socar_api.manage import ManageResource

class CarType(ManageResource, BaseResource):
    model = CarTypeModel
    repo = repo(FactCarType, CarTypeModel)  # ORM + Schema → Repository 实例

    class Meta:
        type_ = 'car_types'
        link = '/car_types'
```

### manage.py — 自动 CRUD 校验链

``ManageResource`` 为后台管理类资源提供开箱即用的 CRUD 操作链：

| 方法 | 执行链 |
|------|--------|
| `post()` | `verify_duplicate → verify_rel → add_user → handle_request_data → repo.create_one` |
| `patch()` | `verify_rel → verify_exist → parse_patch_data → handle_request_data → repo.update_one` |
| `delete()` | `verify_exist → verify_auth → verify_use → add_user → repo.delete_one` |

### db_middleware — 请求级 Session

`setup_app()` 已自动注册 `db_middleware`，无需手动添加：

```python
# setup_app() 内部等价于：
from fastapi.middleware.base import BaseHTTPMiddleware
from socar_api.data.session import db_middleware
app.add_middleware(BaseHTTPMiddleware, dispatch=db_middleware)
```

每个请求自动创建所有已注册数据库的 session；请求成功时调用 `on_commit`，异常时调用 `on_rollback`，最终统一调用 `on_close`。资源内通过 `self.db` 访问（默认 `db_mysql`）。

### 数据库扩展机制（SessionProvider） <a id="session-provider"></a>

框架通过 `SessionProvider` 协议支持接入任意数据库类型（MongoDB、Redis 等），不需要修改框架代码。核心契约只有四个生命周期方法：

```python
from socar_api.data.session import _db, SessionProvider

class SessionProvider(Protocol):
    def create(self): ...          # 创建会话实例
    def on_commit(self, session): ...    # 请求成功时调用
    def on_rollback(self, session): ...  # 请求异常时调用
    def on_close(self, session): ...     # 请求结束时调用（无论成败）
```

接入方实现 Provider 并通过 `_db.register(name, provider)` 注册，此后 `db_middleware` 和 `BatchSessionManager` 即可统一管理该数据库的生命周期。

**示例：接入 MongoDB**

```python
from pymongo import MongoClient
from socar_api.data.session import _db, SessionProvider

class MongoProvider(SessionProvider):
    """全局连接池 — create() 复用同一个 client，on_close() 空操作。"""
    def __init__(self, url, dbname, **kw):
        self._client = MongoClient(url, **kw)
        self._dbname = dbname

    def create(self):
        return self._client[self._dbname]

    def on_commit(self, db): pass
    def on_rollback(self, db): pass
    def on_close(self, db): pass       # PyMongo 自带连接池，全局复用不关闭

_db.register('db_mongo', MongoProvider('mongodb://localhost:27017', 'myapp'))
```

Resource 中按现有方式使用：

```python
class UploadFileResource(BaseResource):
    session_type = 'mysql'         # self.db       → MySQL

    async def post(self, *args, **kwargs):
        self.curd().create_one(db=self.db, ...)    # self.db      → MySQL
        save_file(db=self.request.state.db_mongo)  # db_mongo     → MongoDB
```

## GET 批量聚合（Batch API）

`setup_app()` 自动注册 `POST /_batch` 端点，将多个 GET 子请求合并为单次 HTTP 调用。

### 请求格式

```json
POST /_batch
{
  "operations": [
    "/books",
    "/books/42",
    "/authors?filter[name][op]=ct&filter[name][value]=张",
    "/books?include=author"
  ],
  "mode": "concurrent",
  "format": "detailed",
  "request_timeout": 30,
  "max_urls": 100
}
```

| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `operations` | `list[str]` | 必填 | 子请求 URL 列表 |
| `mode` | `"concurrent"` / `"sequential"` | `"concurrent"` | 并发或串行执行 |
| `format` | `"detailed"` / `"default"` | `"default"` | 响应格式 |
| `request_timeout` | `int` | `30` | 单个子请求超时（秒） |
| `max_urls` | `int` | `100` | 单次请求最多 URL 数 |

### 响应格式

**default** — 所有子请求成功才返回，否则整体报错：

```json
{
  "articles": [...],
  "authors": [...],
  "books": {...}
}
```

**detailed** — 部分失败不中断，分别返回结果和错误：

```json
{
  "results": {
    "/books": { "status": 200, "body": [...] },
    "/books/42": { "status": 200, "body": {...} }
  },
  "errors": {
    "/authors?filter...": { "status": 500, "detail": "..." }
  }
}
```

### 架构要点

- 每个子请求创建**独立的 DB session**，避免并发竞态
- 子请求鉴权由各路由自身 `Auth` 处理，batch 端点鉴权使用 `_get_batch_auth()`
- 使用 Starlette `path_regex` 做路由匹配，封装在框架内部，下游资源无需感知

## 文档

- [框架概述 & 核心概念](docs/index.md)
- [API 参考](docs/api.md)
- [原子操作 & 声明式权限 & 文件上传](docs/guide.md) — 高级主题
- [原子操作设计文档](docs/atomic.md)

## 安装指定版本

```bash
pip install socar-api==0.7.13
```

---

[更新日志](CHANGELOG.md) 包含完整的版本历史。
