Fastapiorm：高效便捷的Python非同步ORM框架

Fastapiorm是一款基於asyncio和pydantic的Python非同步ORM框架，為fastapi應用程序提供支持。與其他ORM框架相比，它不僅提供快速、高效、安全和易於使用的方式來管理資料庫，還可以使用原生的Python非同步語法進行操作。

一、使用Fastapiorm進行Model類重命名

1、在開發的應用程序中，Model類的命名可能受到一些限制。例如，類名必須唯一、不能與其他類重複等。Fastapiorm提供了一種方便的方式來重命名Model類，以滿足應用程序的命名需求。下面是一個示例代碼：


from fastapiorm import Model, StringField

class User(Model):
    name: str = StringField(max_length=50, min_length=1)

    class Config:
        orm_mode = True
        alias = "user_info"

在這個示例中，我們使用了Fastapiorm提供的Config配置屬性，為User Model類重命名為user_info。這使得我們可以避免Model類命名衝突，確保代碼的可維護性。

2、除了為Model類進行重命名外，Config還可以設置其他參數，例如orm_mode，這樣可以在JSON序列化和反序列化期間獲得更好的性能。下面是一個示例代碼：


from fastapiorm import Model, StringField

class User(Model):
    name: str = StringField(max_length=50, min_length=1)

    class Config:
        orm_mode = True
        alias = "user_info"

在這個示例中，我們設置了orm_mode為True，以便在查詢資料庫時可以更快地反序列化JSON數據。

二、Fastapiorm的CRUD操作

Fastapiorm支持一個非常簡單的API來管理資料庫，無需編寫任何SQL查詢。下面是一些常見的CRUD操作示例：

1、創建數據：


from fastapi import FastAPI
from fastapiorm import create_database, Database, Model, StringField

database_url = "sqlite:///./data.db"

app = FastAPI()

class User(Model):
    name: str = StringField(max_length=50, min_length=1)

database = Database(database_url)

@app.on_event("startup")
async def startup():
    await database.connect()
    await create_database(database)

@app.post("/user")
async def create_user(user: User):
    await user.save()

    return {"msg": "user created!"}

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

在這個示例中，我們首先定義了一個名為User的Model類，用於管理用戶信息。然後，在啟動事件時，我們通過調用create_database函數創建了資料庫。

在創建數據時，我們只需要使用Model類來創建一個可序列化的Python對象，然後使用它的save方法來保存到資料庫中。

2、讀取數據：


@app.get("/user/{user_id}")
async def get_user(user_id: int):
    user = await User.filter(id=user_id).first().values()
    if not user:
        return {"msg": "user not found!"}

    return user[0]

在這個示例中，我們定義了一個名為get_user的路由，它可以根據用戶提供的id查詢用戶信息。在查詢時，我們使用了User類的filter方法，並傳入id參數來篩選數據。然後調用first返回符合條件的第一條數據，最後使用values函數獲取所有數據，並返回符合條件的第一個。

3、更新數據：


@app.put("/user/{user_id}")
async def update_user(user_id: int, user: User):
    result = await User.filter(id=user_id).update(**user.dict(exclude_unset=True))
    if not result:
        return {"msg": "user not found!"}

    return {"msg": "user updated!"}

在這個示例中，我們定義了一個名為update_user的路由，用於更新用戶信息。使用User類的filter方法來查找要更新的用戶，然後調用update方法來更新用戶信息。在更新時，我們使用exclude_unset=True來排除用戶未提供的欄位值。最後返回更新結果。

4、刪除數據：


@app.delete("/user/{user_id}")
async def delete_user(user_id: int):
    result = await User.filter(id=user_id).delete()
    if not result:
        return {"msg": "user not found!"}

    return {"msg": "user deleted!"}

在這個示例中，我們定義了一個名為delete_user的路由，用於刪除用戶信息。使用User類的filter方法來查找要刪除的用戶，然後調用delete方法來刪除用戶信息。最後返回刪除結果。

三、Fastapiorm的連接池管理

Fastapiorm提供了一個內置的非同步資料庫連接池，可以自動管理HTTP請求期間的資料庫連接。這個連接池可以被多個HTTP請求使用，而不必擔心出現資料庫連接不足的問題。下面是一個連接池管理示例：


from fastapiorm import create_pool, close_pool, Database

database_url = "sqlite:///./data.db"

database = Database(database_url)

async def startup():
    await database.connect()
    await create_pool(database)

async def shutdown():
    await close_pool(database)
    await database.disconnect()

在這個示例中，我們定義了一個名為database的Database對象，並在啟動時調用create_pool函數來創建一個連接池管理器。在關閉應用程序時，我們調用close_pool函數來關閉連接池管理器，以便釋放所有連接。在startup和shutdown事件中調用這些函數，可以確保在應用程序運行期間一直處於活動狀態。

四、Fastapiorm的事務管理

Fastapiorm提供了一個簡單的API用於管理事務。下面是一個示例代碼：


from fastapiorm import Database, transaction

database_url = "sqlite:///./data.db"

database = Database(database_url)

async def startup():
    await database.connect()

async def shutdown():
    await database.disconnect()

@app.post("/user/transaction")
async def create_user_transaction(user: User):
    async with transaction():
        await user.save()

    return {"msg": "user created!"}

在這個示例中，我們定義了一個名為async with transaction()的上下文管理器，用於管理事務。在管理事務時，我們通過調用user.save()方法來保存用戶信息。如果任何異常拋出，則上下文管理器將自動回滾事務。

五、Fastapiorm的欄位類型

Fastapiorm提供了一些基本的pydantic欄位類型，用於定義Model類中的屬性。下面是一些常見的欄位類型：

1、整數類型：IntegerField

2、浮點數類型：FloatField

3、布爾類型：BooleanField

4、字元串類型：StringField

5、日期類型：DateField

6、時間類型：TimeField

7、日期時間類型：DateTimeField

使用這些欄位類型可以自動執行數據類型驗證和轉換，以確保插入資料庫的數據類型正確。

六、Fastapiorm的查詢構建器

Fastapiorm提供了一種靈活的API來構建複雜的查詢語句。下面是一個示例代碼：


from fastapiorm import Database, Model, StringField

database_url = "sqlite:///./data.db"

database = Database(database_url)

class User(Model):
    name: str = StringField(max_length=50, min_length=1)

async def startup():
    await database.connect()

async def shutdown():
    await database.disconnect()

@app.get("/user/search")
async def search_user(name: str):
    query = User.select().where(User.name.contains(name))
    users = await query.values()

    return users

在這個示例中，我們定義了一個名為search_user的路由，用於根據用戶提供的name查詢用戶信息。使用User類的select方法來獲取一個查詢構建器。然後使用where方法來添加查詢條件，使用.contains方法查詢name欄位的包含關係。最後使用values函數獲取查詢結果。

七、Fastapiorm的數據遷移

Fastapiorm還提供了一個數據遷移工具，可以用於管理資料庫模式的版本控制。下面是一個示例代碼：


from fastapiorm.migrations import Migrations

migrations = Migrations("./migrations", url=database_url)

@migrations.create(name="users")
def create_users_table(connection):
    with connection.cursor() as cursor:
        cursor.execute("CREATE TABLE IF NOT EXISTS users (id INT PRIMARY KEY, name TEXT)")

@migrations.upgrade(name="users", version=1, to_version=2)
def upgrade_users_table(connection):
    with connection.cursor() as cursor:
        cursor.execute("ALTER TABLE users ADD COLUMN age INT")

在這個示例中，我們定義了一個名為migrations的Migrations對象，並使用它來管理資料庫遷移。使用create方法創建一個名為users的數據表。使用upgrade方法，在版本1和版本2之間升級users表，並添加一個名為age的新列。

八、Fastapiorm的拓展

Fastapiorm支持使用其他拓展來增強其API和功能。例如，我們可以使用fastapi-pagination擴展來在FastAPI應用程序中實現分頁功能：


from fastapi import FastAPI
from fastapi_pagination import Page, paginate
from fastapiorm import Database, Model, StringField

database_url = "sqlite:///./data.db"

database = Database(database_url)

class User(Model):
    name: str = StringField(max_length=50, min_length=1)

async def startup():
    await database.connect()

async def shutdown():
    await database.disconnect()

app = FastAPI()

@app.get("/users")
async def get_users():
    query = User.all()
    return await paginate(query)

在這個示例中，我們在FastAPI應用程序中使用fastapi-pagination擴展來添加分頁支持。在路由函數中，我們使用Fastapiorm的all方法來獲取所有User數據，並使用paginate函數將其分頁返回。

九、總結

Fastapiorm是一款高效便捷的Python非同步ORM框架。它提供了快速、高效、安全和易於使用的API來管理資料庫，並使用原生的Python非同步語法進行操作。與其他ORM框架相比，它具有更好的性能和更簡潔的代碼結構。使用Fastapiorm，我們可以輕鬆地構建高質量的Python應用程序，從資料庫管理到數據遷移，從欄位類型到分頁支持，無所不包。

原創文章，作者：小藍，如若轉載，請註明出處：https://www.506064.com/zh-tw/n/186499.html