socar-datakit 0.1.4

轻量级数据访问框架 — Repository/Filter/Mapper/Session/Cache

socar-datakit

轻量级数据访问框架，提供从 HTTP 请求参数到 SQL 查询的完整链路能力。

PyPI:   pip install socar-datakit
Import: import socar_datakit
Python: >= 3.11

定位

做什么

HTTP 参数 (filter/sort/page)                    SQL / ClickHouse
       │                                            ▲
       ▼                                            │
   ┌──────────┐    ┌────────────┐    ┌──────────────┐
   │  Filter  │───→│ Repository │───→│  SchemaMapper │
   │  /Sort   │    │  .get_multi│    │  .from_orm()  │
   └──────────┘    │  .get_one  │    └──────────────┘
                   │  .create   │
                   │  .update   │    ┌──────────────┐
                   │  .delete   │    │  DBSession   │
                   └────────────┘    │  .new_mysql()│
                                     │  .new_ch()   │
                                     └──────────────┘

不做什么

不做	理由
不提供 HTTP 路由注册	那是 socar_api 的职责
不提供 JSON:API 序列化	那是 socar_api 的职责
不提供认证鉴权	那是 socar_api + 认证服务的职责
不提供业务逻辑	那是项目 Resource 层的职责
不依赖 socar_api	通过 duck typing 与 socar_api 的 Filter 类型结构兼容

设计原则

1. 零 project 内部依赖 — 不 import models / db / api / socar_api
2. DB-agnostic — 同一套 API 操作 MySQL 和 ClickHouse
3. 组合优于继承 — repo() 工厂函数一行创建 CRUD
4. Pydantic v2 原生 — 使用 model_validate(from_attributes=True)
5. 请求级 Session — 通过 FastAPI 中间件注入，不用类级单例

---

快速开始

安装

# 核心（MySQL + SQLAlchemy）
pip install socar-datakit

# 全功能（MySQL + ClickHouse + Redis 缓存）
pip install socar-datakit[clickhouse,cache]

最小示例

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
from socar_datakit import repo, init_db
from socar_datakit.session import db_middleware
from orm.dim_car import DimCar
from models.car import CarsModel

app = FastAPI()

# 1. 启动时初始化连接池
init_db(conf)

# 2. 请求级 Session 中间件
app.add_middleware(BaseHTTPMiddleware, dispatch=db_middleware)

# 3. 一行创建 Repository
CarRepo = repo(DimCar, CarsModel)

# 4. 在 Resource 中使用
cars = CarRepo.get_multi(db=request.state.db_mysql, skip=0, limit=20)

---

核心 API

Repository

from socar_datakit import repo

# 工厂函数
CarRepo = repo(DimCar, CarsModel)

# ── 查询 ──
cars = CarRepo.get_multi(db=session, filters=filter_obj, skip=0, limit=20)
car  = CarRepo.get_one(db=session, id_=42)
total = CarRepo.count(db=session, filters=filter_obj)
exists = CarRepo.exists(db=session, filters=filter_obj)

# ── 写入（MySQL 事务表）──
new_user = CarRepo.create_one(db=session, data=user_data)
updated  = CarRepo.update_one(db=session, id_=1, data=update_data)
deleted  = CarRepo.delete_one(db=session, id_=1)

# ── 批量 ──
users = CarRepo.create_batch(db=session, datas=[...])           # safe 模式（默认）
users = CarRepo.create_batch(db=session, datas=[...], batch_mode='fast')  # 大批量
count = CarRepo.delete_batch(db=session, ids=[1, 2, 3])

Repository 是 DB-agnostic 的 — 底层用 MySQL 还是 ClickHouse 取决于传入的 db：

DB 类型	session 来源	典型用途
MySQL	request.state.db_mysql	用户、收藏、反馈等事务数据
ClickHouse	request.state.db_clickhouse	销量、价格、热度等分析数据

get_multi 参数说明：

参数	类型	默认值	说明
db	sqlalchemy.orm.Session	必传	数据库 session
filters	Filter | FilterAnd | FilterOr | None	None	过滤条件树
skip	int	0	分页偏移（对应 JSON:API page[offset]）
limit	int | None	100	分页大小（None 表示不限制）
sorts	list[Sort] | None	None	排序条件
fields	list[str] | None	None	稀疏字段集（JSON:API sparse fieldsets）

Filter 系统

from socar_datakit import Filter, FilterAnd, FilterOr

# 等值过滤
Filter(field='brand', op='eq', value='BMW')

# eq 传 list 自动转为 IN 查询
Filter(field='brand', op='eq', value=['BMW', 'BYD'])

# 范围过滤
Filter(field='price', op='bt', value=(100000, 200000))

# 多值过滤
Filter(field='city', op='in', value=['北京', '上海', '广州'])

# 组合过滤（两种风格均可）
FilterAnd(Filter(field='brand', op='eq', value='BMW'),
          Filter(field='year', op='gte', value=2024))

FilterOr(Filter(field='brand', op='eq', value='BMW'),
         Filter(field='brand', op='eq', value='BYD'))

# 时间范围快捷方式
from socar_datakit import time_range
Filter(field='time', op='bt', value=time_range(1700000000, 1735689599))

支持的操作符：

操作符	含义	示例
eq	等于 / 多值转 IN	Filter('brand', 'eq', 'BMW')
ne	不等于 / 多值转 NOT IN	Filter('brand', 'ne', 'BMW') / Filter('brand', 'ne', ['BMW', 'BYD'])
gt	大于	Filter('price', 'gt', 100000)
gte	大于等于	Filter('price', 'gte', 100000)
lt	小于	Filter('price', 'lt', 200000)
lte	小于等于	Filter('price', 'lte', 200000)
ct	包含（contains）	Filter('name', 'ct', '宝马')
nct	不包含	Filter('name', 'nct', '宝马')
bw	以...开头	Filter('name', 'bw', '华')
ew	以...结尾	Filter('name', 'ew', '有限公司')
bt	范围（between）	Filter('price', 'bt', (100000, 200000))
in	集合成员	Filter('city', 'in', ['北京', '上海'])
nin	非集合成员	Filter('city', 'nin', ['北京'])
nu	为空（is null）	Filter('name', 'nu', None)
nn	非空（is not null）	Filter('name', 'nn', None)
aeq	数组相等	Filter('id', 'aeq', [1, 2])
act	数组包含	Filter('tags', 'act', '热门')

兼容性： socar_datakit.Filter 与 socar_api.query.Filter 结构完全一致（field / op / value），通过 duck typing 互通。项目 Resource 层负责桥接。

Sort 系统

from socar_datakit import resolve_sort

# 定义排序条件（duck typing，只需 .field + .asc 属性）
sorts = [{'field': 'sales', 'direction': 'desc'}]
order_by = resolve_sort(orm=FactCarSale, sorts=sorts)
# → [FactCarSale.sales.desc()]

# 多字段排序
sorts = [
    type('Sort', (), {'field': 'brand', 'asc': True})(),
    type('Sort', (), {'field': 'sales', 'asc': False})(),
]

SchemaMapper

from socar_datakit import SchemaMapper

mapper = SchemaMapper(CarsModel)

# ORM → Schema（Pydantic v2 model_validate）
schema = mapper.from_orm(orm_instance)

# 批量
schema_list = mapper.multi_from_orm([orm_1, orm_2])

# Schema → 字典（用于 INSERT / UPDATE，默认排除 id + None 值）
data = mapper.to_orm_dict(schema_instance)

# Schema → ORM 实例
orm_obj = mapper.to_orm_instance(schema_instance, DimCar)

内部基于 Pydantic v2 model_validate(obj, from_attributes=True)，要求 Schema 配置 model_config = ConfigDict(from_attributes=True)。

Session 管理

from socar_datakit import DBSession, db_middleware, init_db

# 1. 启动时初始化（通常在 main.py）
init_db(conf)
# 或指定数据库配置节（ClickHouse 可选，不传则跳过）
init_db(conf, mysql_db='ads_v4')
init_db(conf, mysql_db='ads_v4', clickhouse_db='')  # 仅 MySQL

# 2. 注册请求级中间件
from starlette.middleware.base import BaseHTTPMiddleware
app.add_middleware(BaseHTTPMiddleware, dispatch=db_middleware)

# 3. 在 Resource 中获取 session
db_mysql      = getattr(request.state, 'db_mysql', None)
db_clickhouse = getattr(request.state, 'db_clickhouse', None)

# 4. 或手动创建 session（不在请求上下文中时）
mysql = DBSession().new_mysql()
ch    = DBSession().new_clickhouse()

中间件自动处理：commit（成功）→ rollback（异常）→ close（始终）。

缓存

from socar_datakit.cache import RedisCache, redis_flushdb

# 定义缓存类
class CarsCache(RedisCache):
    prefix = 'cars'       # hash key 前缀
    ttl = 300             # 过期时间（秒）

# Hash 操作
CarsCache.hash_set('url:xxx', {'data': 'value'})
data = CarsCache.hash_get('url:xxx')

# Key-Value 操作
CarsCache.set('key', value)
value = CarsCache.get('key')

# 版本号缓存失效
redis_flushdb(db=clickhouse_session, version_query=get_version_fn)
# 60 秒内版本未变化则跳过 flush（内存 TTL）

工具

from socar_datakit import uuidgen

# 22 位密码学安全短 UUID（128 位熵）
uid = uuidgen()   # → "K8mQn2xR9p..."

# 异常
from socar_datakit.errors import NotFoundError, DuplicateEntryError, ConflictError

raise NotFoundError(resource='Car', id=42)
raise DuplicateEntryError(detail='品牌名称已存在')
raise ConflictError(detail='数据冲突')

自定义列类型

from socar_datakit.types import (
    UUID,         # 字符串 UUID ↔ bytes
    TsDateTime,   # int 时间戳 ↔ DATETIME（UTC+8）
    TsDate,       # int 时间戳 ↔ DATE（UTC+8）
    Array,        # 字符串数组 ↔ Python list
    CKDate32,     # int 时间戳 ↔ ClickHouse Date32（UTC+8）
)

---

与 socar_api 的关系

socar-datakit                    socar_api
─────────────                    ─────────
Repository CRUD                  路由注册 (register_routes)
Filter/Sort 解析                 请求参数解析 (ArgsModel)
SchemaMapper 转换                JSON:API 序列化 (serializer)
DBSession 会话管理               Security/Scope 鉴权
RedisCache 缓存                  文件上传/下载基类
                                 文档自动生成 (OpenAPI)

两者职责正交，不互相依赖。socar_datakit 的 Filter 类型通过 duck typing 与 socar_api.filter.Filter 结构兼容，Resource 层在中间负责桥接。

---

依赖

核心（始终安装）

依赖	最低版本	用途
pydantic	>=2.0	Schema 模型基类
sqlalchemy	>=2.0	ORM 核心

可选

依赖	extras 标签	用途
clickhouse-sqlalchemy >=0.3.0	[clickhouse]	ClickHouse 方言
clickhouse-driver >=0.2.0	[clickhouse]	ClickHouse 驱动
redis >=5.0	[cache]	缓存后端

---

开发

# 安装开发环境
cd socar-datakit
python -m venv .venv
source .venv/bin/activate
pip install -e ".[clickhouse,cache]"
pip install pytest

# 运行测试
python -m pytest tests/ -v

# 构建
pip install build
python -m build

# 发布到 PyPI
pip install twine
twine upload dist/*

包结构

socar-datakit/
├── pyproject.toml
├── README.md
├── src/
│   └── socar_datakit/
│       ├── __init__.py      # 公开 API 导出
│       ├── repository.py    # Repository 类 + repo() 工厂
│       ├── filter.py        # Filter 树 → SQL WHERE
│       ├── sort.py          # Sort 列表 → SQL ORDER BY
│       ├── mapper.py        # ORM ↔ Pydantic Schema
│       ├── session.py       # DBSession + db_middleware + init_db
│       ├── spec.py          # Filter/FilterAnd/FilterOr 类型定义
│       ├── types.py         # SQLAlchemy 自定义列类型
│       ├── errors.py        # 异常定义
│       ├── uuidgen.py       # 短 UUID 生成
│       └── cache/
│           ├── __init__.py  # 导出 RedisCache / redis_flushdb
│           ├── redis.py     # RedisCache 基类
│           └── versioned.py # 版本号缓存控制 + 内存 TTL
└── tests/
    ├── conftest.py
    └── test_unit.py         # 72 个单元测试