ns-orm 0.0.2

A ORM built on Pydantic models.

ns-orm

一个参考 Django ORM 设计的轻量级 ORM（不依赖 SQLAlchemy），以 Pydantic 作为模型层，支持同步/异步、CRUD、关联预取与最小建表能力。

数据库连接/连接池由 lesscode-database 负责，ns-orm 仅消费“执行器”来执行 SQL。

更多功能与完整用法见 doc/index.md。

安装

pip install ns-orm

数据库驱动与连接方式由 lesscode-database 决定。

快速开始（同步）

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 ForeignKey, Int, Model, 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")
Post.objects.create(user_id=u.id, title="hello")

posts = Post.objects.prefetch_related("user").all()
print(posts[0].user.name)

connect, _connect_info = getattr(db_options, "default")
if not hasattr(connect, "cursor") and hasattr(connect, "connection"):
    connect = connect.connection()

qs = User.objects.filter(id=u.id)
sql, params = qs._select_sql()
prepared = db.dialect.prepare(sql, params)
cursor = connect.cursor()
cursor.execute(prepared.sql, prepared.params)
row = cursor.fetchone()
print(row)

快速开始（异步）

import asyncio

from lesscode_database.connection_info import ConnectionInfo
from lesscode_database.db_options import db_options

from ns_orm import acreate_all, get_connection


async def main():
    db_options.conn_list = [
        ConnectionInfo(dialect="sqlite3", name="default_async", dsn="test_async.db", async_enable=True),
    ]

    db = get_connection("default_async")
    await acreate_all(db, [User, Post])

    u = await User.objects.using("default_async").acreate(name="alice")
    posts = await (
        Post.objects.using("default_async").filter(user_id=u.id).prefetch_related("user").all()
    )
    print(posts[0].user.name)


asyncio.run(main())

迁移命令

见 doc/migrations.md。

打包与发布

见 doc/release.md。