结合fastapi-users与Langserve轻松实现大语言接口用户认证

在做大模型开发的过程中,相信很多小伙伴都是对大模型开发感兴趣,却对 fastapi 这个框架并不熟悉,但是,实际开发的项目确需要用户鉴权,这时候就会很头疼,查阅官方文档发现,官方虽然有例子,但是写的非常简单,只有一个自定义的verify_token函数,函数内容还需要自己实现,如果投入学习成本在 fastapi 的用户认证上就显得得不偿失。而 fastapi 是有现成的用户认证的框架的,比如 fastapi-users。今天,我就带领大家完成一个使用 fastapi-users来实现带有用户认证功能的Langserve接口。

以上是官方的认证鉴权的方法,实现过程需要自己来完成。

实现步骤:

下载fastapi-users

pip install 'fastapi-users[sqlalchemy]'

代码目录结构

创建 langserve 项目的方法我就不再介绍了,感兴趣的同学,可以看我的这篇公众号文章:LangServe全面使用指南

.
├── app
│   ├── db.py
│   ├── __init__.py
│   ├── __pycache__
│   ├── schemas.py
│   ├── server.py
│   └── users.py
├── Dockerfile
├── packages
│   └── README.md
├── poetry.lock
├── pyproject.toml
├── README.md
└── test.db

其中,db.py、users.py、schemas.py都是新建的 python 文件,这个目录结构也是fastapi-users推荐的拆分方式。

代码编写

db.py

from typing import AsyncGeneratorfrom fastapi import Depends
from fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyUserDatabase
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBaseDATABASE_URL = "sqlite+aiosqlite:///./test.db"class Base(DeclarativeBase):passclass User(SQLAlchemyBaseUserTableUUID, Base):passengine = create_async_engine(DATABASE_URL)
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)async def create_db_and_tables():async with engine.begin() as conn:await conn.run_sync(Base.metadata.create_all)async def get_async_session() -> AsyncGenerator[AsyncSession, None]:async with async_session_maker() as session:yield sessionasync def get_user_db(session: AsyncSession = Depends(get_async_session)):yield SQLAlchemyUserDatabase(session, User)

schemas.py

import uuidfrom fastapi_users import schemasclass UserRead(schemas.BaseUser[uuid.UUID]):passclass UserCreate(schemas.BaseUserCreate):passclass UserUpdate(schemas.BaseUserUpdate):pass

users.py

import uuid
from typing import Optionalfrom fastapi import Depends, Request
from fastapi_users import BaseUserManager, FastAPIUsers, UUIDIDMixin
from fastapi_users.authentication import (AuthenticationBackend,BearerTransport,JWTStrategy,
)
from fastapi_users.db import SQLAlchemyUserDatabasefrom app.db import User, get_user_dbSECRET = "SECRET"class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):reset_password_token_secret = SECRETverification_token_secret = SECRETasync def on_after_register(self, user: User, request: Optional[Request] = None):print(f"User {user.id} has registered.")async def on_after_forgot_password(self, user: User, token: str, request: Optional[Request] = None):print(f"User {user.id} has forgot their password. Reset token: {token}")async def on_after_request_verify(self, user: User, token: str, request: Optional[Request] = None):print(f"Verification requested for user {user.id}. Verification token: {token}")async def get_user_manager(user_db: SQLAlchemyUserDatabase = Depends(get_user_db)):yield UserManager(user_db)bearer_transport = BearerTransport(tokenUrl="auth/jwt/login")def get_jwt_strategy() -> JWTStrategy:return JWTStrategy(secret=SECRET, lifetime_seconds=3600)auth_backend = AuthenticationBackend(name="jwt",transport=bearer_transport,get_strategy=get_jwt_strategy,
)fastapi_users = FastAPIUsers[User, uuid.UUID](get_user_manager, [auth_backend])current_active_user = fastapi_users.current_user(active=True)

注意: 这些都是fastapi-users官方用例部分,最主要的是 server.py中的结合部分。

from fastapi import FastAPI, Depends
from fastapi.responses import RedirectResponse
from langserve import add_routesfrom contextlib import asynccontextmanagerfrom app.db import User, create_db_and_tables
from app.schemas import UserCreate, UserRead, UserUpdate
from app.users import auth_backend, current_active_user, fastapi_users
from langchain.chat_models import ChatOpenAI
from langchain_community.chat_models.moonshot import MoonshotChat@asynccontextmanager
async def lifespan(app: FastAPI):# Not needed if you setup a migration system like Alembicawait create_db_and_tables()yieldapp = FastAPI(lifespan=lifespan,#dependencies=[Depends(current_active_user)]
)app.include_router(fastapi_users.get_auth_router(auth_backend), prefix="/auth/jwt", tags=["auth"]
)
app.include_router(fastapi_users.get_register_router(UserRead, UserCreate),prefix="/auth",tags=["auth"],
)
app.include_router(fastapi_users.get_reset_password_router(),prefix="/auth",tags=["auth"],
)
app.include_router(fastapi_users.get_verify_router(UserRead),prefix="/auth",tags=["auth"],
)
app.include_router(fastapi_users.get_users_router(UserRead, UserUpdate),prefix="/users",tags=["users"],
)@app.get("/authenticated-route")
async def authenticated_route(user: User = Depends(current_active_user)):return {"message": f"Hello {user.email}!"}add_routes(app,MoonshotChat(),path="/openai",dependencies=[Depends(current_active_user)],
)@app.get("/")
async def redirect_root_to_docs():return RedirectResponse("/docs")# Edit this to add the chain you want to add
# add_routes(app, NotImplemented)if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)

这里,我用的是kimi的api接口,最主要的改变是在add_routes中增加了dependencies参数,引入了fastapi-users封装好的current_active_user。

add_routes(app,MoonshotChat(),path="/openai",dependencies=[Depends(current_active_user)], #这句代码是关键
)

最后,看下实际调用 invoke 接口后的效果吧!只需要在调用的接口的 header 中加入Authorization即可,对应的值是bearer加 token。这个 token 是调用 login 接口后返回的。

调用login接口获取 token

headers 中加入token

调用 invoke 后的效果

总结

怎么样?是不是很简单?如果有不理解的地方或者需要本代码的,欢迎与我联系。

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

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

uni-app的地图定位与距离测算功能的实现

文章目录 一、引言二、uni-app地图定位实现三、距离测算技术四、完整代码五、结论本文着重探讨了如何在uni-app中实现地图定位,以及如何计算当前定位与目标位置之间的距离。 一、引言 在移动应用开发中,地图定位与距离测算是常见的功能需求。无论是出行导航、位置签到,还是…

学习笔记之——3DGS-SLAM系列代码解读

最近对一系列基于3D Gaussian Splatting(3DGS)SLAM的工作的源码进行了测试与解读。为此写下本博客mark一下所有的源码解读以及对应的代码配置与测试记录~ 其中工作1~5的原理解读见博客: 学习笔记之——3D Gaussian Splatting及其在SLAM与自动…

ios包上架系列 二、Xcode打应用市场ipa包

打包的时候一定要断开网络&#xff0c;上线包名只能在打包机配置 检查是否是正式环境&#xff0c;先在模拟器上运行 1、版本名称和本号号记得在这里更改&#xff0c;否则不生效 原因 &#xff1a;info.list <string>$(FLUTTER_BUILD_NAME)</string><key>CFB…

mysql查看数据库表容量大小

【推荐】单表行数超过 500 万行或者单表容量超过 2GB&#xff0c;才推荐进行分库分表。 说明&#xff1a;如果预计三年后的数据量根本达不到这个级别&#xff0c;请不要在创建表时就分库分表。 1. 查询所有数据库记录数和容量 SELECTtable_schema AS 数据库,SUM(table_rows) …

贪心算法|763.划分字母区间

力扣题目链接 class Solution { public:vector<int> partitionLabels(string S) {int hash[27] {0}; // i为字符&#xff0c;hash[i]为字符出现的最后位置for (int i 0; i < S.size(); i) { // 统计每一个字符最后出现的位置hash[S[i] - a] i;}vector<int> …

jenkins 启动linux节点时 控制台中文显示问号乱码

新增一个jenkins节点时&#xff0c;遇到了控制台中文输出问号的问题。 网上各种配置jenkins的全局变量&#xff0c;都不行。 最终是 节点列表 ->对应节点 -> 启动方式 -> 高级 添加JVM选项 -Dfile.encodingUTF-8

图像处理-采样方法概述

在图像处理中&#xff0c;上采样&#xff08;upsampling&#xff09;和下采样&#xff08;downsampling&#xff09;是常用的操作&#xff0c;用于改变图像的分辨率或大小。这些方法通常用于图像处理任务&#xff0c;如图像放大、缩小、压缩等。下面是一些常见的上采样和下采样…

MySQL基础练习题:习题16-20

这部分主要是为了帮助大家回忆回忆MySQL的基本语法&#xff0c;数据库来自于MySQL的官方简化版&#xff0c;题目也是网上非常流行的35题。这些基础习题基本可以涵盖面试中需要现场写SQL的问题。 列出部门名称和这些部门的员工信息&#xff0c;同时列出没有员工的部门 唯一要注…

面试题:Rabbitmq怎么保证消息的可靠性?

面试题&#xff1a;Rabbitmq怎么保证消息的可靠性? RabbitMQ作为一款广泛使用的消息中间件&#xff0c;提供了多种机制来确保消息的可靠性。要保证消息的可靠性&#xff0c;需要在消息的生产和消费过程中考虑以下几个关键方面&#xff1a; 1. 消息持久化 生产者端持久化 发…

Python 序列化与反序列化

目录 1、基本概念 2、JSON模块 2.1、dumps() 与 loads() 函数 2.2、dump() 与 load() 函数 2.3、bool 、None 类型的序列化与反序列化 3、pickle模块 3.1、dumps() 与 loads() 函数 3.2、dump() 与 load() 函数 1、基本概念 说明&#xff1a;通过文件操作&#xff0c;…

个人博客系统项目(SpringBoot+Linux部署上线)

在学完SpringBoot框架、MyBatis后&#xff0c;直接开始做第一个项目&#xff1a;博客系统 首先&#xff0c;该博客系统包含核心功能有&#xff1a; 一、登录、注册、退出登录功能。 二、没有登陆前可以查看博客首页以及博客展示的分页处理&#xff0c;以及点击查看博客可以…

Restful API接口规范(以Django为例)

Restful API接口规范(以Django为例) Restful API的接口架构风格中制定了一些规范&#xff0c;极大的简化了前后端对接的时间&#xff0c;以及增加了开发效率 安全性保证–使用https路径中带 api标识路径中带版本号数据即资源&#xff0c;通常使用名词操作请求方式决定操作资源…

浅谈网络安全威胁与防御策略

企业网络安全威胁概述 外部威胁&#xff1a;来自网络安全威胁&#xff0c;比如DDOS攻击&#xff0c;病毒&#xff0c;sql注入&#xff0c;木马&#xff0c;蠕虫&#xff0c;等网络入侵&#xff0c;网络扫描&#xff0c;垃圾邮件&#xff0c;钓鱼邮件&#xff0c;针对web的攻击…

集合系列(十九) -List与数组互转操作介绍

一、数组转List 方式一 通过 Arrays.asList(strArray) 方式&#xff0c;将数组转换List&#xff0c;但不能对List增删&#xff0c;只能查改&#xff0c;否则抛异常&#xff0c;例子如下&#xff1a; public static void main(String[] args) {//创建一个字符串数组String[] …

小程序中配置scss

找到&#xff1a;project.config.json 文件 setting 模块下添加&#xff1a; "useCompilerPlugins": ["sass","其他的样式类型"] 配置完成后&#xff0c;重启开发工具&#xff0c;并新建文件 结果&#xff1a;

微信小程序自定义关闭按钮在弹窗下面的效果

效果图: 我之前用vant 的popup的弹窗写&#xff0c;会出现close图标移动到弹窗内容外部不可见。 自定义代码&#xff1a; popup.JS/*** 生命周期函数--监听页面初次渲染完成*/onReady() {//自定义弹窗 动态获取屏幕高度var that this;wx.getSystemInfo({success: (result) &…

centos7 开启防火墙关闭防火墙开机自启

centos7 开启防火墙永久 在CentOS 7中&#xff0c;可以使用firewalld来管理防火墙。以下是启用并设置防火墙为永久的步骤&#xff1a; 启动firewalld服务&#xff1a; sudo systemctl start firewalld 设置firewalld服务为开机自启&#xff1a; sudo systemctl enable firew…

探探各个微前端框架

本文作者为 360 奇舞团前端开发工程师 微前端架构是为了在解决单体应用在一个相对长的时间跨度下&#xff0c;由于参与的人员、团队的增多、变迁&#xff0c;从一个普通应用演变成一个巨石应用(Frontend Monolith)后&#xff0c;随之而来的应用不可维护的问题。这类问题在企业级…

【ELFK】Filebeat+ELK 部署

FilebeatELK 部署 Node1节点&#xff08;2C/4G&#xff09;&#xff1a;node1/192.168.67.11 Elasticsearch Kibana Node2节点&#xff08;2C/4G&#xff09;&#xff1a;node2/192.168.67.12 Elasticsearch Apache节点&#xff1a;apache/192.168.67.10 …

linux学习:队列

目录 顺序队列 管理结构体设计 初始化 入队 出队 链式队列 例子 用户输入整数则入队&#xff0c;字母则出队 我们对队列里的元素操作是有严格限制的&#xff1a;插入一个新节点&#xff0c;必须插入到 指定的一端&#xff0c;而删除一个已有节点&#xff0c;则必须在另…