【Fastapi框架】Fastapi的使用和进阶

news2024/9/28 13:23:24

原文作者：我辈李想
版权声明：文章原创，转载时请务必加上原文超链接、作者信息和本声明。

文章目录

  • 前言
  • 一、项目启动
    • 1.安装
    • 2.示例
    • 3.启动
    • 4.路由
  • 二、进阶
    • 1.请求数据
    • 2.静态文件加载
    • 3.路由管理
    • 4.跨域配置
    • 5.自定义中间件
    • 6.使用jwt中间件

前言

一、项目启动

1.安装

pip install fastapi  uvicorn

2.示例

新建main.py文件，

from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}

3.启动

方式一：命令启动

uvicorn main:app --reload

在这里插入图片描述

方式二:在mian.py中加启动代码

# mian.py

if __name__ == '__main__':
    uvicorn.run(
        app=app,
        host="127.0.0.1",
        port=5678
    )

4.路由

http://127.0.0.1:8000/
http://127.0.0.1:8000/items/1
http://127.0.0.1:8000/items/1?q=2
http://127.0.0.1:5678/docs#/
http://127.0.0.1:5678/redoc

二、进阶

1.请求数据

1.get请求:路径参数+查询参数

路由：http://127.0.0.1:8000/items/1?q=2，其中1是路径参数，q=2是查询参数。

@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}

2.post/put/delete请求：路径参数+请求体

路由：http://127.0.0.1:8000/items/1，其中1是路径参数，item是请求体。pydantic 可以验证输入和输出。

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    is_offer: Union[bool, None] = None

@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_name": item.name, "item_id": item_id}

3.嵌套数据

# 1.定义基础模型
class Image(BaseModel):
    url: str

# 2.赋给其他类属性
class Item(BaseModel):
    image: Union[Image, None] = None  # 单个
    images: Union[List[Image], None] = None  # 多个

4.文件上传

from fastapi import FastAPI, status, File, Form, UploadFile

app = FastAPI()

# 方法一：上传的文件会存在内存中，适合小型文件 
@app.post("/api/uploadFile/action")
async def create_file(file: Annotated[bytes, File()]):  
	writeBytes('./media',file)
	return {
        'code':200,
        "msg":'success'
    }

# 方法二：UploadFile
@app.post("/api/uploadFile/action")
async def create_upload_file(file: UploadFile):
	writeBytes('./media',file)
	return {
        'code':200,
        "msg":'success'
    }

# 上传多个文件
@app.post("/api/uploadFile/action")
async def create_upload_files(files: list[UploadFile]):
	writeBytes('./media',file)
	return {
        'code':200,
        "msg":'success'
    }
    
# 用 File() 添加额外设定
@app.post("/api/uploadFile/action")
async def create_upload_files(
    files: Annotated[
        list[UploadFile], File(description="Multiple files as UploadFile")
    ],
):
	writeBytes('./media',file)
	return {
        'code':200,
        "msg":'success'
    }
    
# 将file写入dirs目录文件
def writeBytes(dirs,file):
    bytesFile=file.file.read()
    filename=file.filename
    if not os.path.exists(dirs):
        os.makedirs(dirs)
    with open(dirs+'/'+ filename, "wb") as f:
        f.write(bytesFile)

5.数据校验

使用fastapi.params库中的函数，FastAPI会自动验证请求参数。Query，Bath 等。

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(q: str = Query(None, min_length=3, max_length=50)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

6.request参数

@router.api_route("/test/{apiname}", methods=['GET', 'POST'])
async def test1(apiname, request: Request):
    args = await request.form()  # form数据
    args2 = await request.json()   # json数据
    print(apiname)
    print(request.method)
    return {'result': '这是一个GET或者POST'}

7.文档注释

@router.post('/roll-plan/kpi/', summary='获取输入数据', description='', tags=['算法'])
async def get_roll_plan(times: Annotated[RollPlanTimeItem, Body(examples=[{"start_time": "2023-12-12 00:00:00",
                                                                           "end_time": "2023-12-13 00:00:00"}],
                                                                )]
                        ):

8.请求体对象转json

from datetime import datetime
from typing import Union

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    title: str
    timestamp: datetime
    description: Union[str, None] = None

@app.put("/items/{id}")
def update_item(id: str, item: Item):
    json_compatible_item_data = jsonable_encoder(item)
    # item.json()
    # item.dict()
    fake_db[id] = json_compatible_item_data

9.sqlalchemy查询数据转json

roll_plans = session.query(rollPlan.AppRollPlanEntity).filter(
	rollPlan.AppRollPlanEntity.StartTime >= time_range[0],
	rollPlan.AppRollPlanEntity.StartTime <= time_range[1]).all()
	
# 方式一
item_data = [result[0].__dict__. for result in roll_plans]  # 将每个结果转换成字典形式
json_data = json.dumps(serialized_data)  # 将字典序列化为JSON字符串
# 方式二
from sqlalchemy.ext.serializer import loads, dumps
# 序列化
serialized_data = dumps(obj)

# 反序列化
deserialized_data = loads(serialized_data)
user = User(**deserialized_data)

2.静态文件加载

from fastapi import FastAPI
from starlette.staticfiles import StaticFiles

app = FastAPI()

# 静态文件配置使用mount挂载 第一个参数为应用的前缀，第二个为路径，第三个为别名
app.mount('/static', StaticFiles(directory='static'), name='static')

这里需要再main.py的同级目录创建static文件，在html文件中需要使用url_for引入静态文件。

<link rel="stylesheet" href="{{ url_for('static',path='demo.css') }}">

3.路由管理

1.在maipn.py的同级创建user文件夹,新建userRouter.py文件

from typing import Union

from fastapi import FastAPI, APIRouter, Request

router = APIRouter(prefix='/bp', tags=['bp'])

@router.get('/')
def read_root():
    return {"Hello": "World"}

@router.get('/items/{item_id}')
async def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}


# @router.get('/test/{apiname}')
# @router.post("/test/{apiname}")
@router.api_route("/test/{apiname}", methods=['GET', 'POST'])
async def test1(apiname, request: Request):
	args = await request.form()
    print(apiname)
    print(request.method)
    return {'result': '这是一个GET或者POST'}

2.修改mapin.py内容

from fastapi import FastAPI
from user import userRouter 

app = FastAPI()
app.include_router(userRouter.router)

if __name__ == '__main__':
    uvicorn.run(
        app=app,
        host="127.0.0.1",
        port=5678
    )

4.跨域配置

# 跨域配置
origins = [
    "http://localhost:3000",
]
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

5.自定义中间件

1.已有中间件

HTTPSRedirectMiddleware是fasapi自带的中间件

app.add_middleware(HTTPSRedirectMiddleware)
app.add_middleware(
    TrustedHostMiddleware, allowed_hosts=["tigeriaf.com", "*.tigeriaf.com"]
)

2.自定义中间件

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

6.使用jwt中间件

在 FastAPI 中使用 JWT 中间件，可以通过自定义 FastAPI 的中间件来实现。以下是在 FastAPI 中使用 JWT 中间件的步骤。

  1. 安装 PyJWT。
pip install PyJWT
  1. 导入 PyJWT 库和 FastAPI 组件。
import jwt
from fastapi import FastAPI, HTTPException
from fastapi.middleware import Middleware
from fastapi.middleware.cors import CORSMiddleware
from starlette.requests import Request
from starlette.status import HTTP_401_UNAUTHORIZED
  1. 定义一些全局变量，例如：
SECRET_KEY = "mysecretkey"  # 生产环境中应使用更复杂的密钥
ALGORITHM = "HS256"  # 哈希算法
ACCESS_TOKEN_EXPIRE_MINUTES = 30  # 访问令牌的过期时间，单位为分钟
  1. 定义 JWT 认证的中间件函数。
async def auth_middleware(request: Request, call_next):
    try:
        token: str = request.headers["Authorization"].split()[1]
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials")
        request.state.user = username
    except (KeyError, IndexError, jwt.JWTError):
        raise HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials")
    response = await call_next(request)
    return response

auth_middleware 函数获取请求头中的访问令牌，并解码 JWT。如果 JWT 解码成功，则将用户名添加到请求对象的 state 属性中。如果 JWT 解码失败，则抛出 HTTPException 异常。

  1. 在 FastAPI 中添加中间件。
middleware = [
    Middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]),
    Middleware(auth_middleware)
]

app = FastAPI(
    title="My API",
    openapi_url="/api/v1/openapi.json",
    docs_url="/api/v1/docs",
    redoc_url=None,
    middleware=middleware
)

在 FastAPI 的构造函数中，使用 Middleware 类添加 CORSMiddleware 和 auth_middleware 中间件。auth_middleware 中间件将在每个路由处理函数执行前进行验证。因此，只需将路由的处理函数包装在 async def 函数中即可。

@app.get("/protected")
async def protected_route(request: Request):
    current_user = request.state.user
    return {"message": "Hello, {}".format(current_user)}

在以上示例中，使用 Request 类作为处理函数的一个参数，然后从请求的 state 属性中获取用户。

以上是使用 JWT 中间件的步骤，你可以根据项目的需求对上述代码进行进一步修改和优化。

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

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

相关文章

轻松学习视频剪辑：视频转码批量转为序列图片技巧，从新手到高手

轻松学习视频剪辑：视频转码批量转为序列图片技巧，从新手到高手

随着视频剪辑的普及&#xff0c;越来越多的爱好者开始尝试自己动手进行视频剪辑。对于新手来说&#xff0c;掌握一些实用的技巧是很有必要的。在视频剪辑中&#xff0c;有时候要将视频片段转换为序列图片&#xff0c;以便进行更灵活的编辑和处理。本文详解云炫AI智剪的一种简单…
阅读更多...
Python爬虫实战(基础篇)—9获取某个城市天气(附完整代码)

Python爬虫实战(基础篇)—9获取某个城市天气(附完整代码)

文章目录 专栏导读背景1、网址(请求URL)2、查看请求方法、参数3、请求初步测试代码请求成功 4、数据清洗(lxmlxpath)【城市实时天气】【日期】lxmlxpath 代码【室外温度】【体感温度】lxmlxpath 代码【天晴情况】【全天气温】lxmlxpath 代码【气压值】【降水量】【风力情况】【…
阅读更多...
Ransac 算法的探索和应用

Ransac 算法的探索和应用

Ransac 算法python 应用和实现 Ransac 算法是一种常用的图像匹配算法&#xff0c;在参数估计领域也经常被使用到。针对估计各种曲线的鲁棒模型参数&#xff0c;效果显著。这里对ransac算法进行某些探索。 python program: import numpy as np import matplotlib.pyplot as p…
阅读更多...
UniGui使用CSS移动端按钮标题垂直

UniGui使用CSS移动端按钮标题垂直

unigui移动端中按钮拉窄以后&#xff0c;标题无法垂直居中&#xff0c;是因为标题有一个padding属性&#xff0c;在四周撑开一段距离。会变成这样&#xff1a; 解决方法&#xff0c;用css修改padding&#xff0c;具体做法如下 首先给button的cls创建一个cls,例如 然后添加css&…
阅读更多...
python numpy 两种方法将相同shape的一维数组合并为二维数组

python numpy 两种方法将相同shape的一维数组合并为二维数组

1 np.column_stack 最简单的一种方法 将多个一维数据按【列】合并为二维数组 import numpy as np# a b 都是一维数组 a np.array((1,2,3)) b np.array((2,3,4))# 变成二维 merge np.column_stack((a,b)) # array([[1, 2],[2, 3],[3, 4]])2 np.hstack 尽管该函数也是对【列…
阅读更多...
桂电|《操作系统》实验一：UNIX/LINUX及其使用环境（实验报告）

桂电|《操作系统》实验一：UNIX/LINUX及其使用环境（实验报告）

桂林电子科技大学2023-2024学年 第 一 学期 操作系统A 实验报告 实验名称 实验一 UNIX/LINUX及其使用环境 实验指导老师&#xff1a; 成绩 院 系 计算机与信息安全学院 专业 计算机科学与技术(卓越工程) 学 号 姓名 课内序…
阅读更多...
多表查询、事务、索引

多表查询、事务、索引

目录 数据准备 分类 内连接 外连接 子查询 事务 四大特性 索引 数据准备 SQL脚本&#xff1a; #建议&#xff1a;创建新的数据库 create database db04; use db04;-- 部门表 create table tb_dept (id int unsigned primary key auto_increment comment 主键…
阅读更多...
物流实时数仓：数仓搭建（DWD）一

物流实时数仓：数仓搭建（DWD）一

系列文章目录 物流实时数仓&#xff1a;采集通道搭建 物流实时数仓&#xff1a;数仓搭建 物流实时数仓&#xff1a;数仓搭建&#xff08;DIM&#xff09; 物流实时数仓&#xff1a;数仓搭建&#xff08;DWD&#xff09;一 文章目录 系列文章目录前言一、文件编写1.目录创建2.b…
阅读更多...
leetcode 162. 寻找峰值（优质解法）

leetcode 162. 寻找峰值（优质解法）

代码&#xff1a; class Solution {public int findPeakElement(int[] nums) {int left0,rightnums.length-1;while (left<right){int midleft(right-left)/2;if(nums[mid]>nums[mid1]){rightmid;}else {leftmid1;}}return left;} } 题解&#xff1a; 通过题意进行分析…
阅读更多...
CRM是怎样帮助企业从激烈竞争中脱颖而出的？

CRM是怎样帮助企业从激烈竞争中脱颖而出的？

有限的市场机会与资源推动了市场竞争。市场竞争是在所难免的&#xff0c;但企业可以借助CRM管理系统调整其业务策略&#xff0c;在市场上很多竞争者中突围。CRM系统怎样帮助企业赢得市场竞争&#xff1f; 以下五个功能点是关键&#xff1a;数据分析、客户管理、合作伙伴、营销自…
阅读更多...
XCP详解「4.2·问题-加载信号过多导致FIFO buffer overflow」

XCP详解「4.2·问题-加载信号过多导致FIFO buffer overflow」

APE write报问题 报文也提示80 溢出 检查测量配置
阅读更多...
【噪音控制 】 铁氧体磁珠

【噪音控制 】 铁氧体磁珠

1. 片状铁氧体磁珠的直流重叠特性 片状铁氧体磁珠是一种使用铁氧体的电感器。因此&#xff0c;当大电流通过时&#xff0c;需要特别注意由于磁饱和所造成的性能改变。图1是电流通过片状铁氧体磁珠时的阻抗值的变化示例。 图1 片状铁氧体磁珠的直流重叠特性示例 正因如此&am…
阅读更多...
2023一起益企广东省中小企业数字化赋能活动（深圳站）成功举办

2023一起益企广东省中小企业数字化赋能活动（深圳站）成功举办

12月12日&#xff0c;由广东工业和信息化厅指导&#xff0c;广东省中小企业服务中心、深圳市中小企业服务局主办&#xff0c;深圳联通承办的2023年“一起益企”广东省中小企业数字化赋能专项对接志愿服务活动&#xff08;深圳站&#xff09;在深圳成功举办。 本次活动涵盖中小企…
阅读更多...
Failed to open the referenced table ‘qrtz_job_details‘

Failed to open the referenced table ‘qrtz_job_details‘

依赖【表】或者【其他对象】执行拉到最前面即可。
阅读更多...
RocketMQ 总体概括

RocketMQ 总体概括

目录 概述RocketMQ 领域模型MQ 解决的问题电商平台案例初步设计引入中间件设计 MQ 选型结束 概述 官网地址 RocketMQ 领域模型 官方领域模型概述 下面图&#xff0c;是在自己理解的基础上&#xff0c;对官方的模型图添加了一些。 Topic&#xff1a;主题&#xff0c;可以理解…
阅读更多...
用栈解决迷宫问题

用栈解决迷宫问题

思想 使用栈来解决迷宫问题的思想是通过深度优先搜索算法来探索迷宫中的路径。栈的特点是后进先出&#xff0c;这正好符合深度优先搜索的思想&#xff0c;即先探索一个方向直到无法继续为止&#xff0c;然后回溯到上一个节点&#xff0c;再探索其他方向。 具体来说&#xff0…
阅读更多...
DC电源模块：为您的电子设备提供稳定可靠的能量

DC电源模块：为您的电子设备提供稳定可靠的能量

DC电源模块&#xff1a;为您的电子设备提供稳定可靠的能量 BOSHIDA DC电源模块是一种电子设备&#xff0c;用于为其他电子设备提供稳定可靠的直流电能。它通常由一个输入端&#xff0c;一个输出端和一些内部电路组成。输入端通常接收来自交流电源或其他电源的电能&#xff0c;经…
阅读更多...
前端走向未来：真相还是焦虑的贩卖？

前端走向未来：真相还是焦虑的贩卖？

目录 一、为什么会出现“前端已死”的言论 二、你如何看待“前端已死” 三、前端技术的未来发展趋势 四、前端人&#xff0c;该如何打好这场职位突围战&#xff1f; 我的其他博客 一、为什么会出现“前端已死”的言论 近来&#xff0c;IT圈内流传着“Java 已死、前端已凉”…
阅读更多...
冷链托盘四向穿梭车｜适用于-18℃～-25℃海格里斯HEGERLS冷库型托盘四向车系统

冷链托盘四向穿梭车｜适用于-18℃～-25℃海格里斯HEGERLS冷库型托盘四向车系统

近年来随着物流行业的迅猛发展&#xff0c;托盘式四向穿梭车在电力、食品、医用、冷链等等行业得到了广泛应用&#xff0c;尤其在冷链物流场景中应用较多&#xff0c;目前设备已具备在-20℃至-25℃的环境中运行&#xff0c;尤其是-18℃及以下的冷链系统&#xff0c;采用四向穿梭…
阅读更多...
mysql8 windows下修改my.ini配置 this is incompatible with sql_mode=only_full_group_by

mysql8 windows下修改my.ini配置 this is incompatible with sql_mode=only_full_group_by

1、找到安装路径 show variables like %sql_mode;SHOW VARIABLES LIKE config_file;SHOW VARIABLES LIKE %datadir%;SHOW VARIABLES; 2、修改 sql_modeSTRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023