原本的代码

@user_router.post("/register/", summary="用户注册")
def register_user(
        username: str = Body(str, min_length=2, max_length=36),
        password: str = Body(str, min_length=6, max_length=128),
        db: SASession = Depends(get_db),
):
    # 检查用户名是否已存在
    user = db.exec(select(FastZdpUserModel).where(FastZdpUserModel.username == username)).first()
    if user:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名已存在")

    # 创建新用户
    new_user = FastZdpUserModel(username=username, password=hash_256(password))
    db.add(new_user)
    try:
        db.commit()
        db.refresh(new_user)
    except Exception as e:
        print(e)
        db.rollback()
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户注册失败")

    return {"message": "用户注册成功", "user_id": new_user.id}

说明

在fastzdp_sqlmodel基本封装完毕以后，我就决定将其应用到实战中，因为只有经过实际项目的考研，才能够说明框架的稳定性，也更能够在实际使用的过程中发现框架可能存在的一些问题。

原本的入口代码分析

import fastzdp_login
from fastapi import FastAPI
from sqlmodel import SQLModel, Session, create_engine

# 创建数据库引擎
sqlite_url = "mysql+pymysql://root:root@127.0.0.1:3306/fastzdp_login?charset=utf8mb4"
engine = create_engine(sqlite_url, echo=True)

# 确保表存在
SQLModel.metadata.drop_all(engine)
SQLModel.metadata.create_all(engine)

app = FastAPI()

# 伪造一个密钥，实际使用时应该使用安全的方式存储
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

# 令牌有效期
ACCESS_TOKEN_EXPIRE_MINUTES = 30


def get_db():
    db = Session(engine)
    try:
        yield db
    finally:
        db.close()


app.include_router(fastzdp_login.get_user_router(
    get_db,
    SECRET_KEY,
    ALGORITHM,
    ACCESS_TOKEN_EXPIRE_MINUTES * 60
))

if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app, host='0.0.0.0', port=8888)

在这个代码里面，我们需要手动的将get_db方法传递进去，如果使用fastzdp_sqlmodel的话，就只需要传递engine进去就行了。

我们来试试

初步改造如下

@user_router.post("/register/", summary="用户注册")
def register_user(
        username: str = Body(str, min_length=2, max_length=36),
        password: str = Body(str, min_length=6, max_length=128),
):
    # 检查用户名是否已存在
    users = fs.get_by_dict(engine, FastZdpUserModel, {"username": username})
    if users and len(users) > 0:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名已存在")

    # 创建新用户
    new_user = FastZdpUserModel(username=username, password=hash_256(password))
    fs.add(engine, new_user)
    return {}

只不过目前还不能直接测试，因为用户登录接口还没有改造完毕。

接下来，我们继续改造用户登录的接口。

原来的用户登录接口

@user_router.post("/login/", summary="用户登录")
async def login_for_access_token(
        username: str = Body(str, min_length=2, max_length=36),
        password: str = Body(str, min_length=6, max_length=128),
        db: SASession = Depends(get_db),
):
    user = db.exec(select(FastZdpUserModel).where(FastZdpUserModel.username == username)).first()
    if not user:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名或密码错误")
    if not verify_256(password, user.password):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名或密码错误")
    data = {"username": user.username, "id": user.id, "time": time.time(),
            "expired": jwt_token_expired}
    access_token = get_jwt(data, jwt_key, jwt_algorithm)
    return {"access_token": access_token, "token_type": "bearer"}

改造后的用户登录接口

@user_router.post("/login/", summary="用户登录")
async def login_for_access_token(
        username: str = Body(str, min_length=2, max_length=36),
        password: str = Body(str, min_length=6, max_length=128),
):
    # 查找
    users = fs.get_by_dict(engine, FastZdpUserModel, {"username": username})
    if not users or len(users) == 0:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名或密码错误")

    # 校验密码
    user = users[0]
    if not verify_256(password, user.password):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名或密码错误")

    # 生成Token
    data = {
        "username": user.username,
        "id": user.id,
        "time": time.time(),
        "expired": jwt_token_expired,
    }
    access_token = get_jwt(data, jwt_key, jwt_algorithm)

    # 返回
    return {"access_token": access_token, "token_type": "bearer"}

改造入口代码

import fastzdp_login
from fastapi import FastAPI
from sqlmodel import SQLModel, Session, create_engine

# 创建数据库引擎
sqlite_url = "mysql+pymysql://root:root@127.0.0.1:3306/fastzdp_login?charset=utf8mb4"
engine = create_engine(sqlite_url, echo=True)

# 确保表存在
SQLModel.metadata.drop_all(engine)
SQLModel.metadata.create_all(engine)

app = FastAPI()

# 伪造一个密钥，实际使用时应该使用安全的方式存储
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

# 令牌有效期
ACCESS_TOKEN_EXPIRE_MINUTES = 30

app.include_router(fastzdp_login.get_user_router(
    engine,
    SECRET_KEY,
    ALGORITHM,
    ACCESS_TOKEN_EXPIRE_MINUTES * 60
))

if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app, host='0.0.0.0', port=8888)

通过接口文档测试

启动入口代码，访问接口问题。

比较遗憾的是FastAPI自带的接口文档并不稳定，比如今天就碰到了无法访问的问题。

所以我决定使用自己开发的req工具进行接口测试。

测试注册接口：

req -X POST -H 'Content-Type:application/json' -d '{\"username\":\"zhangdapeng\",\"password\":\"zhangdapeng520\"}' http://127.0.0.1:8888/fastzdp_login/register/

可以发现，返回的状态码是200，说明注册成功了。

测试登录接口：

req -X POST -H 'Content-Type:application/json' -d '{\"username\":\"zhangdapeng\",\"password\":\"zhangdapeng520\"}' http://127.0.0.1:8888/fastzdp_login/login/

成功的登录并获取到了JWT Token。

总结

本篇教程主要记录了fastzdp_login通过整合fastzdp_sqlmodel实现版本升级的的详细过程。

升级以后，功能并没有太大的变化，只是底层使用的技术发生了改变。

通过本次实战，也发现了fastzdp_sqlmodel还缺少两个便捷的方法。

• is_exists：用来判断是否存在某条数据
• get_first：获取一条数据

后续我还会继续升级fastzdp_sqlmodel这个框架，敬请关注。

如果你需要本教程的完整源码，或者需要完整的零基础到实战的FastAPI实战录播课，或者需要一对一的私教指导，欢迎留言或者私信我。