fastzdp_login的第一次构建

概述

为了方便能够快捷的实现fastapi实现登录相关功能代码开发，决定开发一个开源的fastapi组件库，想了很多个名字，在检查pypi的时候发现都被占用了，所以最终决定使用fastzdp_login这个名字。
fast代表的时fastapi。zdp代表的是张大鹏。login代表的是登录功能。
希望通过fastzdp_login这个库，能够简化使用fastapi开发登录功能的逻辑，实现低代码低成本的开发。

环境搭建

首先是安装基本依赖：

pip install poetry 
poetry add fastapi

fastapi底层依赖特别多，所以安装特别慢，不信你看：

而我们的zdppy_api框架则是零依赖的，安装速度特别快。所以，如果你zdppy_api和fastapi都会的话，建议你优先使用zdppy_api这个框架。
这里之所以开发fastapi的组件库并开源，主要是为了方便在工作中使用fastapi作为核心框架的同学。

接着，还需要安装操作MySQL数据库的依赖：

poetry add sqlmodel[mysql]

最后，是我们启动服务的依赖：

poetry add uvicorn

配置pip国内源：

pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple

配置poetry国内源：

poetry source add --priority=default mirrors https://pypi.tuna.tsinghua.edu.cn/simple/

用户增删改查示例

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session as SASession
from sqlmodel import SQLModel, Field, Session, create_engine, select
from typing import Optional


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(index=True)
    password: str
    email: str
    phone: str
    avatar: str
    name: str


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

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

app = FastAPI()


# 依赖项，为每个请求提供数据库会话
def get_db():
    db = Session(engine)
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=User)
def create_user(user: User, db: SASession = Depends(get_db)):
    db.add(user)
    db.commit()
    db.refresh(user)
    return user


# 读取所有用户
@app.get("/users/", response_model=list[User])
def read_users(db: SASession = Depends(get_db)):
    return db.exec(select(User)).all()


# 读取单个用户
@app.get("/users/{user_id}", response_model=User)
def read_user(user_id: int, db: SASession = Depends(get_db)):
    user = db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    return user


# 更新用户
@app.put("/users/{user_id}", response_model=User)
def update_user(user_id: int, user: User, db: SASession = Depends(get_db)):
    db_user = db.get(User, user_id)
    if not db_user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    for field in user.__fields__:
        if getattr(user, field.name, None) is not None:
            setattr(db_user, field.name, getattr(user, field.name))
    db.commit()
    db.refresh(db_user)
    return db_user


# 删除用户
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int, db: SASession = Depends(get_db)):
    db_user = db.get(User, user_id)
    if db_user is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    db.delete(db_user)
    db.commit()
    return None


if __name__ == '__main__':
    import uvicorn

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

开发注册接口

from fastapi import FastAPI, Depends, HTTPException, status, Body
from sqlalchemy.orm import Session as SASession
from sqlmodel import SQLModel, Field, Session, create_engine, select
from typing import Optional
from passlib.hash import pbkdf2_sha256 as sha256  # 用于密码哈希


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(index=True)
    password: str


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

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

app = FastAPI()


# 依赖项，为每个请求提供数据库会话
def get_db():
    db = Session(engine)
    try:
        yield db
    finally:
        db.close()


@app.post("/users/register/", status_code=status.HTTP_201_CREATED)
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(User).where(User.username == username)).first()
    if user:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Username already exists")

    # 对密码进行哈希处理
    hashed_password = sha256.hash(password)

    # 创建新用户
    new_user = User(username=username, password=hashed_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="Operation failed")

    return {"message": "User registered successfully", "user_id": new_user.id}

实现登录接口

import time

from fastapi import FastAPI, Depends, HTTPException, status, Body
from sqlalchemy.orm import Session as SASession
from sqlmodel import SQLModel, Field, Session, create_engine, select
from typing import Optional
from passlib.hash import pbkdf2_sha256 as sha256  # 用于密码哈希
from jwt import encode as jwt_encode
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(index=True)
    password: str


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

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

app = FastAPI()

# 创建一个 OAuth2 令牌
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

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

# 令牌有效期
ACCESS_TOKEN_EXPIRE_MINUTES = 30


# 假设你有一个获取用户的函数
def get_user(db: SASession, username: str):
    return db.exec(select(User).where(User.username == username)).first()


# 依赖项，为每个请求提供数据库会话
def get_db():
    db = Session(engine)
    try:
        yield db
    finally:
        db.close()


@app.post("/users/register/", status_code=status.HTTP_201_CREATED)
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(User).where(User.username == username)).first()
    if user:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Username already exists")

    # 对密码进行哈希处理
    hashed_password = sha256.hash(password)

    # 创建新用户
    new_user = User(username=username, password=hashed_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="Operation failed")

    return {"message": "User registered successfully", "user_id": new_user.id}


# 登录接口
@app.post("/token/", response_model=dict, status_code=status.HTTP_200_OK)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: SASession = Depends(get_db)):
    user = get_user(db, form_data.username)
    if not user:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect username or password")
    if not sha256.verify(form_data.password, user.password):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect username or password")
    access_token = jwt_encode(
        {"username": user.username, "id": user.id, "time": time.time(), "expired": ACCESS_TOKEN_EXPIRE_MINUTES * 60},
        SECRET_KEY,
        algorithm=ALGORITHM,
    ).encode("utf-8")
    return {"access_token": access_token, "token_type": "bearer"}


if __name__ == '__main__':
    import uvicorn

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

后续

后面根据这两个原型接口进行改造。