devbox
小小林熬夜学编程
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

FastAPI--基本操作到部署_uvicorn和fastapi部署

作者：小小林熬夜学编程 | 2024-03-31 13:35:29

赞

踩

uvicorn和fastapi部署

官文敬上:虽然看不明白
官方文档

安装框架

pip install fastapi					-- web框架
pip install uvicorn					--基于高并发的模块,fastapi启动依赖于它

  • 1
  • 2

写一个最简单的接口

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get('/')							#一个get请求
async def root():
    return {'hello':'World'}		    #返回值


if __name__ == '__main__':
    uvicorn.run(app,host='127.0.0.1',port=8000)    #可以直接运行此脚本
  
    
#uvicorn polls:app --reload   
#uvicorn 文件名:实例名 --在代码改变后重启服务器，只能在开发的时候使用
#命令行启动必须进入该文件的根路径

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17

Fastapi自带接口文档

启动项目后访问127.0.0.1:8000/docs
127.0.0.1:8000/redoc

  • 1
  • 2

fastapi使用模板渲染

pip install jinja2		  -- 安装模板渲染模块


import uvicorn
from fastapi import FastAPI
from starlette.requests import Request
from starlette.templating import Jinja2Templates

app = FastAPI()
templates = Jinja2Templates(directory='../templates')     #模板路径


@app.get('/')
async def root(request:Request):   #必须有这个参数,拿不到数据
    return templates.TemplateResponse('index.html',{'request':request,'hello':'测试数据'})
		#渲染的模板名, 后面的hello为模板的变量名,后面是数据

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

# uvicorn polls:app --reload
# uvicorn 文件名:实例名 --在代码改变后重启服务器，只能在开发的时候使用
	



#测试渲染模板index.html中
	
<!DOCTYPE html>
<html lang="en">
    <head>
        <meta charset="UTF-8">
        <title>Title</title>
    </head>
    <body>
        <h1>{{ hello }}</h1>
    </body>
</html>

#启动运行项目.可以看到渲染成功的页面

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40

路径参数的传递

import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.get('/{user_id}/')
async def root(user_id):
    return user_id


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

#请求http://127.0.0.1:8000/12555/      可以看到12555
----------------------------------------------------------------------------












#预定传递路径参数(也就是路径参数必须是我们提前预设的值)
import uvicorn
from enum import Enum
from fastapi import FastAPI

app = FastAPI()

class ModelName(str,Enum):   #意思就是底下{name}中的值必须为我们这个类中定义的三个属性值
    alexent = 'alexnet'
    resnet = 'resnt' 	#比如,我们定义resnt,那么必须在url中输入resnt才可以执行进视图函数
    lenet = 'lenet'		#如果输入为resnet,则不可以进入视图函数        


@app.get('/api/{name}/')		#如果我们name中的值不是我们上面定义好的属性值,那么返回错误json
async def root(name: ModelName):
    if name == ModelName.alexent:				#这种方式检查路径参数比较优雅
        return {'mes':'alexent'}
    if name.value == 'lenet':					#这种方式也可以取到路径参数中的值
        return {'name':name,'mes':'lenet'}
    return {"name":name,'mes':'resnet'}

---------------------------------------------------------------------------------















#路径参数中包含路径参数(例如:api/1/2/3)
@app.get('/api/{name:path}/')
async def root(name):
	return {'mes':name}
#浏览器中随便输入,可接受多个路径参数

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70

参数传递的类型检验(校验)

#python3.6中已经支持声明变量的类型

@app.get('/api/{user_id}/')
async def root(user_id: bool):     #这种情况下user_id被声明为bool类型,如果传递别的参数,那就type error
	return user_id

  • 1
  • 2
  • 3
  • 4
  • 5

查询参数(中间带?的参数)

import uvicorn
from fastapi import FastAPI

app = FastAPI() 
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


@app.get("/items/")					 
async def read_item(page:int=0, size:int=10):   #当然这里不声明的话,在切片的时候声明也是可以的
    return fake_items_db[page : page + size] 
    #默认查10条,从page条开始查
#请求:127.0.0.1:8000/items/?page=1
注意:如果请求127.0.0.1:8000/items/  相当于执行视图函数中参数的默认值

输出为:
[
	{
	item_name: "Bar"
	},
	{
	item_name: "Baz"
	}
]

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23

路径参数和查询参数配合使用

from typing import Optional
import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/api/{name}/")
async def read_item(name:str,age:Optional[str] = None): #官文中这样使用(可以理解为是可选参数)
    if age:											
        return {'name':name,'age':age}
    return name

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

POST请求拿raw(json)数据

from typing import Optional			
from fastapi import FastAPI
from pydantic import BaseModel

class Item(BaseModel):					#必须先定义json参数的模型
    name: str		
    description: Optional[str] = None     #非必选参数
    price: float
    tax: Optional[float] = None				#非必选参数

app = FastAPI()

@app.post("/items/")
async def create_item(item: Item):
    return item
输出如下:

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16

在这里插入图片描述

@app.post("/items/")
async def create_item(item: Item):
	print(item.dict())							#转化字典
    item_dict = item.dict()					#可以通过.对象的方法来获取值
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

POST请求拿from(表单)数据

pip install python-multipart						#装模块

from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):  #可以拿到表单数据
    return {"username": username,"passwod":password}
如下:

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

在这里插入图片描述
文件流上传

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/files/")           #这种方式是在表单中提交,需要前端吧文件转字节码
async def create_file(file: bytes = File(...)):		
    print(file)
    return {"file_size": file}
#注意:这种方式是吧文件存在内存中,适合小文件传输

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9

表单中文件上传

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/files/")
async def upload_file(file: UploadFile = File(...)):
    content = await file.read()					#一次性读取,文件太大服务器就得崩溃
    with open(f'{file.filename}','wb') as f:
        f.write(content)
    return {"filename": file.filename,'content_type':file.content_type}
#注意:这种方式只适合小文件上传
--------------------------------------------------------------------------------











#大文件上传		
from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/files/")
async def upload_file(file: UploadFile = File(...)):
    with open(f'{file.filename}','wb') as f:
        for i in iter(lambda : file.file.read(512),b''):
            f.write(i)
    f.close()

    return {"filename": file.filename,'content_type':file.content_type}

UploadFile
UploadFile 具有以下属性：

filename：str具有上载的原始文件名的A （例如myimage.jpg）。
content_type：str具有内容类型（MIME类型/媒体类型）（例如image/jpeg）的A。
file：一个SpooledTemporaryFile（类似文件的对象）。这是实际的Python文件，您可以将其直接传递给需要“类文件”对象的其他函数或库。
UploadFile具有以下async方法。它们都调用了下面的相应文件方法（使用internal SpooledTemporaryFile）。

write(data)：将data（str或bytes）写入文件。
read(size)：读取size（int）个文件的字节/字符。
seek(offset)：转到文件中的字节位置offset（int）。
例如，await myfile.seek(0)将转到文件的开头。
如果您运行await myfile.read()一次然后需要再次读取内容，则此功能特别有用。
close()：关闭文件。

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51

批量上传文件

from typing import List
from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/file/")
async def create_files(files: List[bytes] = File(...)):
    return {"file_sizes": [len(file) for file in files]} 
#注意:这里必须前端表单中必须为file类型,而不是上面的text类型


@app.post("/files/")
async def create_upload_files(files: List[UploadFile] = File(...)):
    return {"filenames": [file.filename for file in files]}
#注意:表单中包含多个files文件,推荐使用这种方法
----------------------------------------------------------------------------------






#大文件批量上传
@app.post("/files/")
async def create_upload_files(files: List[UploadFile] = File(...)):
    for file in files:
        with open(f'{file.filename}','wb') as f:
            for i in iter(lambda : file.file.read(512),b''):
                f.write(i)
            f.close()
    return {"filenames": [file.filename for file in files]}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31

表单中text与file类型配合使用

from fastapi import FastAPI, File, Form, UploadFile

app = FastAPI()


@app.post("/files/")
async def create_file(
    file: bytes = File(...), fileb: UploadFile = File(...), token: str = Form(...)
):
    return {
        "file_size": len(file),
        "token": token,
        "fileb_content_type": fileb.content_type,
    }

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14

配置跨域资源共享

  • allow_origins-应该允许进行跨域请求的来源列表。例如[‘https://example.org’, ‘https://www.example.org’]。您可以[‘*’]用来允许任何来源。

  • allow_origin_regex-一个正则表达式字符串，与应允许进行跨域请求的原点匹配。例如。‘https://.*.example.org’。

  • allow_methods-跨域请求应允许的HTTP方法列表。默认为[‘GET’]。您可以使用[‘*’]允许所有标准方法。

  • allow_headers-跨域请求应支持的HTTP请求标头列表。默认为[]。您可以[‘*’]用来允许所有标头。的Accept，Accept-Language，Content-Language和Content-Type头总是允许CORS请求。

  • allow_credentials-表示跨域请求应支持cookie。默认为False。

  • expose_headers-指出应该使浏览器可以访问的任何响应头。默认为[]。
    max_age-设置浏览器缓存CORS响应的最长时间（以秒为单位）。默认为600。

      from fastapi.middleware.cors import CORSMiddleware
  
  app = FastAPI()
  
  origins = [
     "*",
  ]
  
  app.add_middleware(
      CORSMiddleware,
      allow_origins=origins,
      allow_credentials=True,
      allow_methods=["*"],
      allow_headers=["*"],
  )

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15

部署fastapi

利用进程管理gunicorn

  • 1

流式响应

在 FastAPI 中，可以通过 StreamingResponse 类来实现流式响应。StreamingResponse 类需要传入一个生成器函数，该函数会在响应被发送到客户端时逐步生成响应内容。

以下是一个示例代码，演示如何使用 StreamingResponse 类返回流式响应：

from fastapi import FastAPI, Response
from fastapi.responses import StreamingResponse
import time

app = FastAPI()

def generate():
    for i in range(10):
        yield f"streaming response {i}\n"
        time.sleep(1)

@app.get("/stream")
async def stream():
    return StreamingResponse(generate())

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14

在上面的示例中，generate 函数是一个生成器函数，它会逐步生成响应内容。stream 函数使用 StreamingResponse 类来返回流式响应，generate 函数作为参数传入 StreamingResponse 类的构造函数中。

当客户端请求 /stream 路径时，stream 函数会返回一个 StreamingResponse 对象，该对象会逐步生成响应内容并发送到客户端。客户端可以通过读取响应内容来逐步接收数据。

需要注意的是，由于 StreamingResponse 类会逐步生成响应内容，因此在生成器函数中需要适当地添加延时，以便客户端有足够的时间来接收数据。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小小林熬夜学编程/article/detail/344755
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号