socar_api

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

简介

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

框架提供两种资源基类：

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

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 集成

安装

pip install socar-api

最低要求: Python 3.12+

第一个资源

1. 定义数据模型

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：

from socar_api import StrictSchemaBase, Field

class BookModel(StrictSchemaBase):
    id: str = Field(None)  # id=1 将抛出 ValidationError

2. 定义资源

from socar_api import BaseResource

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

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

3. 创建应用

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. 启动

uvicorn main:app --reload

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

启动后自动生成：

注意: GET /{link}/{id} 仅在子类覆写了 get() 方法时才会实际注册。若 methods 中包含 GET 但未实现有效的 get()，框架会输出 warning 并跳过该路由。

完整的 CRUD 资源

实现业务逻辑

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 方法中可直接使用，无需额外包裹。

关系资源

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

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

等价于编程式筛选：

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 中。 前端 codegen 可以直接从 paths/{resource}/get/parameters 读取 filter[field] 参数定义。

数据库与 Repository

安装数据库支持：

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

初始化

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 端点
# 注意：必须在所有 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 指定使用的数据库实例：

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 文件：

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)

在资源类中使用：

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 操作链：

db_middleware — 请求级 Session

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

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

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

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

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 中按现有方式使用：

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 调用。

请求格式

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
}

响应格式

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

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

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

{
  "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 做路由匹配，封装在框架内部，下游资源无需感知

文档

• 框架概述 & 核心概念
• API 参考
• 原子操作 & 声明式权限 & 文件上传 — 高级主题
• 原子操作设计文档

安装指定版本

pip install socar-api==0.7.11

更新日志 包含完整的版本历史。