thinkorm を通じてデータベース クエリ ステートメントを最適化して IO 操作を削減する方法

pip install thinkorm

import asyncio
from thinkorm import Model, Field, Database

class User(Model):
    id = Field(int, primary_key=True)
    name = Field(str)

async def main():
    db = Database("sqlite:///:memory:")
    await db.connect()

    # 异步查询
    users = await User.select().where(User.name == "Alice").all()

    for user in users:
        print(user.name)

    await db.disconnect()

asyncio.run(main())

import asyncio
from thinkorm import Model, Field, Database

class User(Model):
    id = Field(int, primary_key=True)
    name = Field(str, index=True)

async def main():
    db = Database("sqlite:///:memory:")
    await db.connect()

    # 使用索引进行查询
    users = await User.select().where(User.name == "Alice").all()

    for user in users:
        print(user.name)

    await db.disconnect()

asyncio.run(main())

import asyncio
from thinkorm import Model, Field, Database

class User(Model):
    id = Field(int, primary_key=True)
    name = Field(str)

async def main():
    db = Database("sqlite:///:memory:")
    await db.connect()

    names = ["Alice", "Bob", "Charlie"]

    # 批量查询
    users = await User.select().where(User.name.in_(names)).all()

    for user in users:
        print(user.name)

    await db.disconnect()

asyncio.run(main())

import asyncio
from thinkorm import Model, Field, Database

class User(Model):
    id = Field(int, primary_key=True)
    name = Field(str)
    age = Field(int)

class Post(Model):
    id = Field(int, primary_key=True)
    content = Field(str)
    user_id = Field(int)

async def main():
    db = Database("sqlite:///:memory:")
    await db.connect()

    # 子查询
    subquery = User.select(User.id).where(User.age > 18)
    posts = await Post.select().where(Post.user_id.in_(subquery)).all()

    for post in posts:
        print(post.content)

    await db.disconnect()

asyncio.run(main())