pgcr-fastapi-orm 1.1.2
为fastapi定制化开发的orm框架
License: MIT
Requires Python: >=3.7,<4.0

# pgcr_fastapi_orm

#### 介绍
为fastapi定制化开发的orm框架

## 快速入门
### 完整增删改查案例
```python
from typing import Optional
import sqlalchemy
from pgcr_fastapi_orm import *

db = Db()
database = db.database
metadata = db.metadata


class BaseMeta(ModelMeta):
    metadata = metadata
    database = database


class Author(Model):
    class Meta(BaseMeta):
        tablename = "authors"

    id: int = Integer(primary_key=True)
    name: str = String(max_length=100)


class Book(Model):
    class Meta(BaseMeta):
        tablename = "books"

    id: int = Integer(primary_key=True)
    author: Optional[Author] = ForeignKey(Author)
    title: str = String(max_length=100)
    year: int = Integer(nullable=True)


engine = sqlalchemy.create_engine(db.DATABASE_URL)
metadata.drop_all(engine)
metadata.create_all(engine)


async def test_add():  # 测试添加
    # 添加一个
    zhangsan = await  db.add(Author, {"name": "张三"})
    print("添加结果: ", zhangsan)
    # 添加外键
    result = await  db.add(Book, {"title": "张三的书", "author": zhangsan, "year": 1998})
    print("添加结果: ", result)
    # 添加多个
    result = await  db.add(Author, [{"name": f"test{i}"} for i in range(1, 100)])
    print("添加结果: ", result)
    # 添加多个外键
    result = await  db.add(Book, [{"title": f"张三的书{i + 1}", "author": zhangsan, "year": 1998} for i in range(1, 100)])
    print("添加结果: ", result)


async def test_find():  # 测试查找
    book = await Book.objects.get(title="张三的书")
    print("查询结果: ", book)
    book = await db.find(Book, {"title": "张三的书"})
    print("查询结果: ", book[0])
    books = await Book.objects.all()
    print("查询结果: ", books)
    books = await db.find(Book)
    print("查询结果: ", books)


async def test_update():  # 测试更新
    # 更新单个
    zhangsan = await Author.objects.get(name="张三")
    zhangsan.name = "张珊珊"
    await zhangsan.update()
    # 使用db更新
    await db.update(Author, {"name": "张三"}, {"name": "李思思"})
    # 复杂条件更新
    await Author.objects.filter(name__contains="张").update(name="李思雨")
    await db.update(Author, {"name__contains": "张"}, {"name": "李思雨"})


async def test_delete():
    book = await Book.objects.get(title="张三的书")
    await book.delete()
    await Book.objects.delete(title="张三的书")
    # 使用db删除
    await db.delete(Book, {"title": "张三的书"})


async def test_join():
    book = await Book.objects.select_related("author").get(title="张三的书")
    print('查询结果: ', book)
    books = await Book.objects.select_related("author").all(title="张三的书")
    print('查询结果: ', books)
    # 使用db查询
    books = await db.join(Book, ['author'], {"title": "张三的书"})
    print('查询结果: ', books)


async def test_page():
    books = await Book.objects.paginate(page=1, page_size=20).all()
    print("查询结果: ", books)
    # 使用db查询
    books = await db.page(Book)
    print("查询结果: ", books)


async def test_agg():
    result = await Book.objects.max(columns=["year"])
    print("查询结果: ", result)
    result = await Book.objects.min(columns=["year"])
    print("查询结果: ", result)
    result = await Book.objects.avg(columns=["year"])
    print("查询结果: ", result)
    result = await Book.objects.sum(columns=["year"])
    print("查询结果: ", result)
    result = await Book.objects.count()
    print("查询结果: ", result)
    # 使用db查询
    print("---")
    result = await db.agg(Book, ["year"], "max")
    print("查询结果: ", result)
    result = await db.agg(Book, ["year"], "min")
    print("查询结果: ", result)
    result = await db.agg(Book, ["year"], "avg")
    print("查询结果: ", result)
    result = await db.agg(Book, ["year"], "sum")
    print("查询结果: ", result)
    result = await db.agg(Book, [], "count")
    print("查询结果: ", result)


async def with_connect(function):
    async with database:
        await function()


import asyncio

asyncio.run(with_connect(test_add))
print("------------------------")
asyncio.run(with_connect(test_find))
print("------------------------")
# asyncio.run(with_connect(test_update))
print("------------------------")
# asyncio.run(with_connect(test_delete))
print("------------------------")
asyncio.run(with_connect(test_join))
print("------------------------")
asyncio.run(with_connect(test_page))
print("------------------------")
asyncio.run(with_connect(test_agg))
metadata.drop_all(engine)
```