ns-orm

受 Django ORM 启发的异步优先 ORM，内置 153 种数据库方言。基于 lesscode-database 提供连接池，支持同步与异步执行。

完整文档见 doc/index.md。

特性

核心

• Django 风格 Model — 字段定义、Meta 配置、自动建表
• QuerySet API — filter/exclude/annotate/order_by/values/select_related/prefetch_related 等 30+ 链式 + 终端方法
• 异步原生 — AsyncQuerySet + Django 兼容 a 前缀别名 (acount/aget/acreate 等)
• 表达式 — F、Q、Case/When、Window、Subquery、Exists、Func、Coalesce、Concat、Cast、Extract、Now
• 聚合 — Count、Sum、Avg、Min、Max、StdDev、Variance
• 关联 — ForeignKey、OneToOneField、ManyToManyField（支持 through 模型）
• 信号 — pre_save、post_save、pre_delete、post_delete、m2m_changed
• 事务 — atomic()、savepoint()、on_commit()
• 验证 — full_clean()、ValidationError、max_length/choices 校验
• 迁移 — 内置迁移系统（CLI）
• Schema — create_all() / acreate_all() 自动建表
• 135 个测试全通过 ✅

数据库

• 153 个数据库方言 — 涵盖标准 RDBMS、云数仓、NoSQL、图、时序、向量、国产数据库
• ClickHouse 专用方言 — 自定义类型映射 + MergeTree DDL
• 注册表架构 — 一行注册新方言、3 种参数风格、2 种标识符引用

快速开始

from typing import Optional
from typing_extensions import Annotated
from lesscode_database.connection_info import ConnectionInfo
from lesscode_database.db_options import db_options
from ns_orm import Count, ForeignKey, Int, Model, Q, String, create_all, get_connection


class User(Model):
    class Meta:
        connect_name = "default"
    id: Annotated[Optional[int], Int(primary_key=True)] = None
    name: Annotated[str, String(max_length=50)]


class Post(Model):
    class Meta:
        connect_name = "default"
    id: Annotated[Optional[int], Int(primary_key=True)] = None
    user_id: Annotated[int, ForeignKey("User", on_delete="CASCADE")]
    title: Annotated[str, String(max_length=200)]


db_options.conn_list = [
    ConnectionInfo(dialect="sqlite3", name="default", dsn="test.db", async_enable=False),
]
db = get_connection("default")
create_all(db, [User, Post])

u = User.objects.create(name="alice")
assert Post.objects.filter(user__name="alice").count() == 1
assert Post.objects.aggregate(total=Count("id"))["total"] == 1
titles = Post.objects.order_by("id").values_list("title", flat=True).all()
posts = Post.objects.prefetch_related("user").all()

异步示例：

import asyncio
from ns_orm import acreate_all, get_connection

async def main():
    db_options.conn_list = [
        ConnectionInfo(dialect="sqlite3", name="default", dsn="test_async.db", async_enable=True),
    ]
    db = get_connection("default")
    await acreate_all(db, [User, Post])
    u = await User.objects.using("default").acreate(name="alice")
    posts = await Post.objects.filter(user_id=u.id).prefetch_related("user").all()

asyncio.run(main())

数据库支持

架构

src/ns_orm/
├── model.py           # Model 基类
├── model_m2m.py       # M2M 管理器
├── database.py        # Database/AsyncDatabase 连接管理
├── fields.py          # 全部字段类型
├── expressions.py     # F/Q/Case/When/Window/Subquery 表达式
├── signals.py         # 信号系统
├── schema.py          # 建表
├── exceptions.py      # 异常
├── query_baseset.py   # BaseQuerySet（链式方法共享基类）
├── query_queryset.py  # QuerySet（同步执行）
├── query_asyncset.py  # AsyncQuerySet（异步执行 + a- 前缀别名）
├── query_state.py     # QueryState/HavingLookup 核心数据结构
├── query_helpers.py   # WHERE/JOIN/聚合编译、验证守卫
├── query_sql.py       # SELECT SQL 构建
├── query_execution.py # 查询执行
├── query_common.py    # 共享辅助函数
├── query_evaluation.py # 求值入口
├── query_prefetch.py  # 预取逻辑
├── query_result.py    # 结果物化
├── query_temporal.py  # 时间截断
├── dialects.py        # 153 个方言注册表
├── cli.py             # 命令行工具
├── migrations/        # 迁移系统
├── vector/            # 向量数据库支持
├── redis/             # Redis 支持
├── graph/             # 图数据库支持
├── mongodb/           # MongoDB 支持
└── elasticsearch/     # Elasticsearch 支持

测试

• 135 个测试 ✅ 全部通过
• 9 个跳过（需要外部依赖：MongoDB, Elasticsearch 等集成测试）
• 覆盖：核心 CRUD、异步、关联、M2M、聚合、表达式、信号、验证、事务、Lookup 类型、方言注册表

许可

MIT License

GitHub · 文档 · 测试 · 方言测试