FastAPI系列：fastapi定制的数据库操作库sqlmodel

官网

sqlmodel

安装

# 安装sqlmodel会自动安装pydantic和sqlalchemy
pip install sqlmodel

使用

# 步骤1，创建sqlmodel引擎
from sqlmodel import create_engine

# driver://用户名:密码@ip/数据库
engine = create_engine("mysql+mysqldb://root:123456@localhost/api")

# 步骤2，定义数据库表映射模型
from typing import Optional
from sqlmodel import Field, SQLModel

class Users(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    nikename: str
    password: str
    email: str

# 步骤3，通过sqlmodel创建逻辑业务表
SQLModel.metadata.create_all(engine)

快速入门之增删改查

from user import Users, engine
from sqlmodel import Field, Session, SQLModel


if __name__  ==  '__main__':
    # 增
    user1 = Users(name='jack', nikename='jj', password='123456', email='[email protected]')
    user2 = Users(name='yamu', nikename='yy', password='123456', email='[email protected]')

    with Session(engine) as session:
        session.add(user1)
        session.add(user2)
        session.commit()
        
    # 查，通过select()
    ## 查询全部数据
    with Session(engine) as session:
        allusers = select(Users)
        results = session.exec(allusers)
        for user in results:
            print(user)
            
    ## 根据条件查询单个数据
    with Session(engine) as session:
        userresult = select(Users).where(Users.name == 'jack')
        # 获取第一条记录
        user = session.exec(userresult).first()
        print(user)
        
    ## 多条件查询
    with Session(engine) as session:
        userresult = select(Users).where(Users.name == 'jack').where(Users.nikename == 'jj')
        # 获取全部记录
        users = session.exec(userresult).all()
        print(users)
        
    ## 另一种多条件查询
    with Session(engine) as session:
        userresult = select(Users).where(Users.name == 'jack', Users.nikename == 'jj')
        # 获取全部记录
        users = session.exec(userresult).all()
        print(users)
        
        
   	# 更新
    with Session(engine) as session:
        # 先查询符合条件的记录
        results = session.exec(select(Users).where(Users.name=='jack'))
        user = results.first()
        user.email  = '[email protected]'
        session.add(user)
        session.commit()
        session.refresh(user)
        
   	## 批量更新  
	with Session(engine) as session:
        # 通过update
        updateusers = update(Users).where(Users.name == 'jack')
        results = session.exec(updateusers.values(email='[email protected]'))
        session.commit()
        
    # 删除
    with Session(engine) as session:
        # 查询符合条件的记录
        user = session.exec(select(Users).where(Users.name == 'jack')).first()
        # 删除该记录
        session.delete(user)
        session.commit()
        
    with Session(engine) as session:
        # 通过delete模块筛选 所有 符合条件的记录，批量进行删除操作
        user = session.exec(delete(Users).where(Users.name == 'jack'))
        session.commit()

异步sqlmodel操作

安装依赖aiomysql

pip install aiomysql

使用

# db/database.py
#  导入异步引擎的模块
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import declarative_base, sessionmaker

# 创建异步引擎对象
async_engine = create_async_engine('mysql+aiomysql://root:[email protected]/api', echo=True)

# 创建orm模型基类
Base = declarative_base()

# 创建异步会话管理对象
sessionLocal = sessionmaker(bind=async_engine, expire_on_commit=False, class_=AsyncSession)

# dependencies/session.py
from sqlalchemy.ext.asyncio import AsyncSession
from typing import AsyncGenerator
from db.database import sessionLocal

async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
    db_session = None
    try:
        db_session = sessionLocal()
        yield db_session
    finally:
        await db_session.close()

# services/user.py
from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession
from models.model import User
from db.database import async_engine, Base
class UserServices:
    @staticmethod
    async def get_user(async_session: AsyncSession, user_id: int):
        result = await async_session.execute(select(User).where(User.id == user_id))
        return result.scalars().first()

# api/user.py
router_user = APIRouter(prefix='/user', tags=['用户'])
@router_user.get('/{user_id}')
async def get(user_id: int, db_session: AsyncSession=Depends(get_db_session)): # 通过依赖注入的方式获取session会话
  result  = await UserServices.get_user(db_session, user_id) # 将session会话对象传递到services方法中
  return {
        'code': 200,
        'msg': 'success',
        'data': {
            'result': result
        }
    }