fastzdp_login的第一次构建

news2024/11/22 21:49:23

概述

为了方便能够快捷的实现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)

后续

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

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1883744.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

CesiumJS【Basic】- #041 绘制纹理线（Entity方式）- 需要自定义着色器

CesiumJS【Basic】- #041 绘制纹理线（Entity方式）- 需要自定义着色器

文章目录 绘制纹理线（Entity方式）- 需要自定义着色器1 目标2 代码2.1 main.ts3 资源文件绘制纹理线（Entity方式）- 需要自定义着色器 1 目标 使用Entity方式绘制纹理线 2 代码 2.1 main.ts import * as Cesium from cesium;const viewer = new Cesium.Viewer
阅读更多...
工程技术类SCI,低分快刊首选期刊，无版面费！

工程技术类SCI,低分快刊首选期刊，无版面费！

1、期刊概况 【期刊简介】IF：1.0-2.0，JCR2区，中科院4区； 【检索情况】SCIE在检 【版面类型】正刊，仅少量版面； 【出刊频率】年刊 2、征稿范围 本刊主要是发表有关能源转型和可再生能源需求相关的研究文…
阅读更多...
windows USB 设备驱动开发-Host端和Device端

windows USB 设备驱动开发-Host端和Device端

Windows 中的 USB 宿主端驱动程序 下图显示了适用于 Windows 的 USB 驱动程序堆栈的体系结构框图。 此图显示了适用于 USB 2.0 和 USB 3.0 的单独 USB 驱动程序堆栈。 当设备连接到 xHCI 控制器时，Windows 加载 USB 3.0 驱动程序堆栈。 Windows 为连接到 EHCI、OHC…
阅读更多...
MySQL进阶：存储过程和函数

MySQL进阶：存储过程和函数

存储过程和函数 1. 简介2. 创建存储过程使用MySQL工作台创建存储过程 3. 删除存储过程4. 参数带默认值的参数参数验证输出参数 5. 变量6. 函数7. 其他约定 1. 简介 存储过程三大作用： 储存和管理SQL代码（置于数据库中，与应用层分离&#xf…
阅读更多...
Redis集群-主从复制、哨兵

Redis集群-主从复制、哨兵

●主从复制：主从复制是高可用Redis的基础，哨兵和集群都是在主从复制基础上实现高可用的。 主从复制主要实现了数据的多机备份，以及对于读操作的负载均衡和简单的故障恢复。缺陷：故障 恢复无法自动化；写操作无法负载均…
阅读更多...
小白学习手册：轻松理解MQ消息队列

小白学习手册：轻松理解MQ消息队列

目录 # 开篇 RabbitMQ介绍 通讯概念 1. 初始MQ及类型 2. MQ的架构 2.1 RabbitMQ的结构和概念 2.2 RabbitMQ消息流示意图 3. MQ下载使用 3.1 Docker下载MQ参考 3.2 进入RabbitMQ # 开篇 MessagesQueue 是一个抽象概念，用于描述消息队列系统的一般特性和功能…
阅读更多...
Ubuntu(通用)—网络加固—ufw+防DNS污染+ARP绑定

Ubuntu(通用)—网络加固—ufw+防DNS污染+ARP绑定

1. ufw sudo ufw default deny incoming sudo ufw deny in from any to any # sudo ufw allow from any to any port 5353 protocol udp sudo ufw enable # 启动开机自启 # sudo ufw reload 更改后的操作2. 防ARP欺骗 华为云教程 arp -d删除dns记录arp -a显示arp表 ipconfi…
阅读更多...
windows USB设备驱动开发-双角色驱动

windows USB设备驱动开发-双角色驱动

在USB的通讯协议中，规定发起连接的一方为主机(Host)，接受连接的一方为设备，这可以用U盘插入电脑举个例子，当U盘插入电脑后，电脑这边主动发起查询和枚举，U盘被动响应查询和数据存取。 USB 双角色驱动程序堆…
阅读更多...
《RepViT Revisiting Mobile CNN From ViT Perspective》

《RepViT Revisiting Mobile CNN From ViT Perspective》

期刊：CVPR 年份：2024 代码：http://https: //github.com/THU-MIG/RepViT 摘要 最近，与轻量级卷积神经网络(CNN)相比，轻量级视觉Transformer(ViTs)在资源受限的移动设备上表现出了更高的性能和更低的延迟。研究人员已…
阅读更多...
大聪明教你学Java | 深入浅出聊 RocketMQ

大聪明教你学Java | 深入浅出聊 RocketMQ

前言 🍊作者简介： 不肯过江东丶，一个来自二线城市的程序员，致力于用“猥琐”办法解决繁琐问题，让复杂的问题变得通俗易懂。 🍊支持作者： 点赞👍、关注💖、留言&#x1f4…
阅读更多...
技术派Spring事件监听机制及原理

技术派Spring事件监听机制及原理

Spring事件监听机制是Spring框架中的一种重要技术，允许组件之间进行松耦合通信。通过使用事件监听机制，应用程序的各个组件可以在其他组件不直接引用的情况下，相互发送和接受消息。 需求 在技术派中有这样一个需求，当发布文章或…
阅读更多...
旋转变压器软件解码simulink仿真

旋转变压器软件解码simulink仿真

1.介绍 旋转变压器是一种精密的位置、速度检测装置，尤其适用于高温、严寒、潮湿、高速、振动等环境恶劣、旋转编码器无法正常工作的场合。旋转变压器在使用时并不能直接提供角度或位置信息，需要特殊的激励信号和解调、计算措施，才能将旋转变压…
阅读更多...
【RT摩拳擦掌】基于RT106L/S语音识别的百度云控制系统

【RT摩拳擦掌】基于RT106L/S语音识别的百度云控制系统

【RT摩拳擦掌】基于RT106L/S语音识别的百度云控制系统 一 文档简介二 平台构建2.1 使用平台2.2 百度智能云2.2.1 物联网核心套件2.2.2 在线语音合成 2.3 playback语音数据准备与烧录2.4 开机语音准备与添加2.5 唤醒词识别词命令准备与添加 三 代码准备3.1 sln-local/2-iot 代码…
阅读更多...
2025第13届常州国际工业装备博览会招商全面启动

2025第13届常州国际工业装备博览会招商全面启动

常州智造 装备中国|2025第13届常州国际工业装备博览会招商全面启动 2025第13届常州国际工业装备博览会将于2025年4月11-13日在常州西太湖国际博览中心盛大举行！目前，各项筹备工作正稳步推进。 60000平米的超大规模、800多家国内外工业装备制造名企将云集…
阅读更多...
Unity Shader 软粒子

Unity Shader 软粒子

Unity Shader 软粒子 前言项目Shader连连看项目渲染管线设置 鸣谢 前言 当场景有点单调的时候，就需要一些粒子点缀，此时软粒子就可以发挥作用了。 使用软粒子与未使用软粒子对比图 项目 Shader连连看 这里插播一点，可以用Vertex Color与…
阅读更多...
KUKA机器人不同运行方式

KUKA机器人不同运行方式

KUKA机器人有以下四种运行方式： 1、手动慢速运行（T1） 2、手动快速运行（T2） 3、自动运行（AUT） 4、外部自动运行（AUT EXT） 将示教器上的钥匙向右旋转，就会…
阅读更多...
【数据结构之B树的讲解】

【数据结构之B树的讲解】

🎥博主：程序员不想YY啊 💫CSDN优质创作者，CSDN实力新星，CSDN博客专家 🤗点赞🎈收藏⭐再看💫养成习惯 ✨希望本文对您有所裨益，如有不足之处，欢迎在评论区提出…
阅读更多...
专题五：Spring源码之初始化容器上下文

专题五：Spring源码之初始化容器上下文

上一篇我们通过如下一段基础代码作为切入点，最终找到核心的处理是refresh方法，从今天开始正式进入refresh方法的解读。 public class Main {public static void main(String[] args) {ApplicationContext context new ClassPathXmlApplicationContext(…
阅读更多...
Study--Oracle-05-Oracler体系结构

Study--Oracle-05-Oracler体系结构

一、oracle 体系概览 Oracle数据库的体系结构通常包括以下主要组件： 1、实例（Instance）：运行数据库的软件环境，包括内存结构（SGA）和进程结构（Background Processes and User Proces…
阅读更多...
Mysql面试合集

Mysql面试合集

概念 是一个开源的关系型数据库。 数据库事务及其特性 事务：是一系列的数据库操作，是数据库应用的基本逻辑单位。 事务特性： （1）原子性：即不可分割性，事务要么全部被执行，要么就…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023