概述
为了方便能够快捷的实现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)
后续
后面根据这两个原型接口进行改造。