一、安装依赖

pip install SQLAlchemy==1.4.46

pip install sqlacodegen

pip install pymysql

二、在根目录新建文件夹db

三、在db文件夹中新建session.py文件

import os
from contextlib import contextmanager
from typing import Generator
from urllib import parse

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

HOST = 'localhost'
PORT = 3306
USERNAME = 'root'
PASSWORD = parse.quote_plus('123456')
DB = 'authcenter'

DB_URI = f'mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DB}'

engine = create_engine(DB_URI, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 这里一定要用with上下文去管理session,否则会出现很多诡异的情况！！！切记

# 创建session
DbSession = sessionmaker(bind=engine)
db_session = DbSession()


@contextmanager
def session_maker(session=db_session):
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()


def get_db() -> Generator:
    """
    get SQLAlchemy session to curd
    :return: SQLAlchemy Session
    """
    db = None
    try:
        db = SessionLocal()
        yield db
    finally:
        if db:
            db.close()


def get_db_connect() -> Generator:
    """
    get SQLAlchemy connect to exec sql
    :return: SQLAlchemy connect
    """
    conn = None
    try:
        conn = engine.connect()
        yield conn
    finally:
        if conn:
            conn.close()


# 逆向工程 自动生成模型文件
if __name__ == '__main__':
    tables = ["user"]
    os.system(
        f'sqlacodegen --tables {",".join(tables)} {DB_URI} > db_models.py')

 温馨提示：在填写PASSWORD的密码栏的时候使用parse.quote_plus('123456')将密码包裹起来。这是防止有些时候密码中包含了特殊字符@，这个和驱动中的@冲突。

四、 执行session.py生成模型文件

五、CRUD

在user文件夹下新建crud文件夹，并在该文件夹下新建user_crud.py文件内容如下：

from sqlalchemy.orm import Session

from db.db_models import User


def getByUsername(db: Session, *, username: str):
    """
    根据用户名查询用户
    :param db: 
    :param username: 
    :return: 
    """
    return db.query(User).filter(User.username == username).first()

六、调用CRUD

修改user文件夹下的views.py如下：

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

from apps.user.crud import user_crud
from apps.user.schemas.user_schemas import Item
from db.session import get_db

router = APIRouter()


@router.post("/login", summary="用户登录", tags=['用户'])
async def login(item: Item, db: Session = Depends(get_db)):
    username = user_crud.getByUsername(db=db, username=item.username)
    return {'message': '登录成功', 'data': username}