socar-api 0.7.1

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

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 扩展支持，批量操作 + 事务回滚
• 文件上传/下载 — 增强的 UploadFileResource / DownloadFileResource 基类，文件类型验证、关系校验、可替换存储后端
• 响应缓存 — @cached() 装饰器一行启用接口缓存，Cache-Aside 模式，可替换后端
• 筛选过滤 — Deep Object 格式，40+ 操作符覆盖字符串/数值/列表/时间/布尔
• 自动模型 — 根据资源自动生成请求/响应模型
• 资源版本 — 内建 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)

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	/books	资源列表
GET	/books/{id}	单个资源
POST	/books	新增资源
PATCH	/books/{id}	更新资源
DELETE	/books/{id}	删除资源
POST	/atomic	原子操作（register_routes() 自动注册）

完整的 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 — 在文章列表中关联作者

查询与筛选

框架自动解析 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.data.session import init_db

# ORM · SQLAlchemy declarative base
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
    pass

# 在应用启动时初始化连接
init_db(
    orm_base=Base,
    mysql_url='mysql+pymysql://user:pass@host/db',
    clickhouse_url='clickhouse://user:pass@host/db',
)

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

方法	执行链
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

from fastapi.middleware.base import BaseHTTPMiddleware
from socar_api.data.session import db_middleware

app.add_middleware(BaseHTTPMiddleware, dispatch=db_middleware)

注册后每个请求自动获取/关闭数据库 session，资源内通过 self.db 访问。

文档

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

安装指定版本

pip install socar-api==0.7.1

---

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