0.SQLModel 是什么

SQLModel 是一个用于从 Python 代码中使用 Python 对象与 SQL 数据库进行交互的库。它被设计为直观、易于使用、高度兼容且健壮。
换而言之，就是将 python 类转化为数据库中数据表及对象的工具
对应的(开发文档)[https://sqlmodel.tiangolo.com/]

1.安装方式

# 前提是在虚拟环境下, 安装sqlmodel以后他会一并安装其他依赖
pip install sqlmodel

2. 使用方式

我们先来模拟一个场景，就拿电动车管理系统来举例
共八张表，分别为用户表(user),角色表(role),权限表(access),电动车表(machine),区域表(area),记录表(record)
用户角色表(userrolelink),角色权限表(roleaccesslink),

models.py

# 引入的包
from sqlmodel import SQLModel, Field, Relationship
from datetime import datetime
from sqlalchemy import JSON, Column
import uuid

# 基础字段
class BasicModel(SQLModel):
    create_by: uuid.UUID = Field(description="创建者")
    create_time: datetime = Field(default=datetime.utcnow(), description="创建时间")
    update_by: uuid.UUID = Field(description="更新者")
    update_time: datetime = Field(default=datetime.utcnow(), description="更新时间")

# 多对多的关系
# table=True,他就代表是数据库中一张表，为false,就是一个Python类
class UserRoleLink(BasicModel, table=True):
    user_id: uuid.UUID = Field(foreign_key="user.user_id", primary_key=True, description="用户ID")
    role_id: uuid.UUID = Field(foreign_key="role.role_id", primary_key=True, description="角色ID")

    user: "User" = Relationship(back_populates="user_role_links")
    role: "Role" = Relationship(back_populates="user_links")

# 一个角色对应多个权限,一个权限可对应多个角色
class RoleAccessLink(BasicModel, table=True):
    role_id: str = Field(foreign_key="role.role_id", primary_key=True, description="角色ID")
    access_id: str = Field(foreign_key="access.access_id", primary_key=True, description="权限ID")

    role: "Role" = Relationship(back_populates="access_links")
    access: "Access" = Relationship(back_populates="access_role_links")


class User(BasicModel, table=True):
    user_id: uuid.UUID = Field(primary_key=True, description="用户ID,用户的唯一标识")
    username: str = Field(description="用户名", index=True)
    password: str = Field(description="加密过后用户的密码")
    user_status: int = Field(default=1, description="用户状态,0为禁用状态,1为可用状态,2表示正在骑行状态", index=True)

    # 附加的属性
    user_role_links: list[UserRoleLink] = Relationship(back_populates="user")

# 一个用户对应多个角色，一个角色对应对个用户
class Role(BasicModel, table=True):
    role_id: str = Field(primary_key=True, description="角色ID,角色的唯一标识")
    role_name: str = Field(description="角色名")
    role_desc: str | None = Field(description="角色描述", default=None)

    user_links: list[UserRoleLink] = Relationship(back_populates="role")
    access_links: list[RoleAccessLink] = Relationship(back_populates="role")

class Access(BasicModel, table=True):
    access_id: str = Field(primary_key=True, description="权限ID,权限的唯一标识")
    access_name: str = Field(description="权限名")
    access_desc: str | None = Field(description="权限描述", default=None)
    access_url: str | None = Field(description="权限URL", default=None)
    parent_id: str | None = Field(description="父亲ID", default=None)
    is_menu: bool = Field(description="是否为菜单", default=False)
    is_verify: bool = Field(description="是否需要验证", default=True)

    access_role_links: list[RoleAccessLink] = Relationship(back_populates="access")

class Machine(BasicModel, table=True):
    machine_id: uuid.UUID = Field(primary_key=True, description="电动车ID,电动车的唯一标识")
    machine_point: dict | None = Field(default=None, description="电动车位置", sa_column=Column(JSON)) # json类型
    machine_battery: int = Field(default=100, description="电动车电量")
    status: int = Field(default=1, description="电动车状态,0为正在骑行中,1为空闲状态,2为损坏,3为正在停止")
    machine_photo: str | None = Field(default=None, description="电动车照片")

    area_id: str = Field(foreign_key="area.area_id", description="区域ID")
    area: "Area" = Relationship(back_populates="machines")
    record: list["Record"] = Relationship(back_populates="machine")


class Area(BasicModel, table=True):
    area_id: str = Field(primary_key=True, description="区域ID,区域的唯一标识")
    area_name: str | None = Field(default=None, description="区域名")
    area_desc: str | None = Field(default=None, description="区域描述")

    machines: list[Machine] = Relationship(back_populates="area")


class Record(BasicModel, table=True):
    record_id: uuid.UUID = Field(primary_key=True, description="记录ID,记录的唯一标识")
    start_time: datetime | None = Field(default=None, description="开始时间")
    end_time: datetime | None = Field(default=datetime.utcnow(), description="结束时间")
    stop_time:int = Field(default=0, description="停车时间")
    consume_battery: int = Field(default=0, description="消耗电量")
    tracejectory: dict | None = Field(default=None, description="轨迹", sa_column=Column(JSON))

    # 附加属性
    user_id: str = Field(foreign_key="user.user_id", description="用户ID")
    user: User = Relationship(back_populates="records")
    machine_id: str = Field(foreign_key="machine.machine_id", description="电动车ID")
    machine: Machine = Relationship(back_populates="records")

mysql 配置文件

mysql.py

# 这里的settings.属性更换成自己的变量即可
from sqlmodel import SQLModel, create_engine
from config import settings
from models.models import User, Role, Machine, Record, Access, UserRoleLink, RoleAccessLink, Area

engine = create_engine(f"mysql+pymysql://{settings.MYSQL_USER}:{settings.MYSQL_PASSWORD}@{settings.MYSQL_IP}:{settings.MYSQL_PORT}/{settings.MYSQL_DATABASE}")


async def register():
    SQLModel.metadata.create_all(engine)

3.启动方式

Events.py

from typing import Callable
from fastapi import FastAPI
from database.mysql import register #这个是上面的mysql配置文件，只是存放可能不同


def startup(app: FastAPI) -> Callable:
    """
    Fastapi启动事件
    :param app: FastAPI
    :return: start_app
    """

    async def app_start() -> None:

        print("------fsray-api start------")
        # 数据库注册
        await register()
        pass
    return app_start


def stopping(app: FastAPI) -> Callable:
    """
    Fastapi停止事件
    :param app: FastAPI
    :return: stop_app
    """

    async def stop_app() -> None:
        print("------fsray-api stop------")

    return stop_app

main.py

from fastapi import FastAPI
from config import settings
from core import Events


application = FastAPI(
    debug=settings.APP_DEBUG, #调试模式
)

# 事件监听
application.add_event_handler("startup", Events.startup(application))
application.add_event_handler("shutdown", Events.stopping(application))


app = application