标签：name 示例 flask fastapi app str import euraka def

flask - fastapi （python 异步API 框架 可以自动生成swagger 文档） 常用示例 以及整合euraka nacos

flask - fastapi    （python 异步API 框架  可以自动生成swagger 文档）  常用示例：

之前使用 flask 需要手动写文档， 这个可以自动生成, 

fastapi 0.82.0
pydantic 1.10.2
python-multipart 0.0.5
uvicorn 0.18.3

 swagger-ui      http://127.0.0.1:5555/docs      

参数可选：

@app.post("/blog/add", tags=["blog"])  # 指定命名空间
async def add_blog_api(content: str = Form(...),
                       operator: str = Form(...),
                       # img: UploadFile = File(...)):
                       img: Optional[UploadFile] = None): # 可选参数
    img_url = None
    if img:
        out_img_dir = compress_img(img.file, quality=20)
        img_url = upload_file(url=UPLOAD_FILE_URL, file_path=out_img_dir)
        print("img url " + str(img_url))
        os.remove(path=out_img_dir)
        print(img_url)
    blog = add_blog(content=content, img=img_url, operator=operator)
    print("post data " + str(blog))
    return str(blog)

 

 

常用示例 (python3.8 )

 pip install fastapi
 pip install uvicorn
 pip install python-multipart
添加跨域：

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

 

# -*- coding:utf-8 -*-
# file_name : base_test.py
# Author : Sea
# Time : 2022-09-06 15:03:12
# pip install fastapi
# pip install uvicorn
# pip install python-multipart
############################################
from typing import Optional, Union

from fastapi import FastAPI, UploadFile, File, Query, Form, Request
import uvicorn
from pydantic import BaseModel

app = FastAPI()


#  ************ get method ***************
@app.get("/")
async def index(xx: str, req: Request):
    print(" 可以通过request 获取一切参数 xx" + str(req.query_params))  # xxxx=13
    print(" 可以通过request 获取一切参数 xx" + str(req.form()))  
    print(" 可以通过request 获取一切参数 xx" + str(req.headers))  
    return {"name": "哈哈哈"}


@app.get("/user/{user_id}")
async def get_user(user_id: Union[int, str], name: Optional[str] = None):
    """通过 Union 来声明一个混合类型，int 在前、str 在后。会先按照 int 解析，解析失败再变成 str
       然后是 name，它表示字符串类型、但默认值为 None（不是字符串），那么应该声明为 Optional[str]"""
    return {"user_id": user_id, "name": name}


# Query 里面除了限制最小长度和最大长度，还有其它的功能
@app.get("/user", tags=["user"])
async def check_length(
        password: str = Query(default="satori", min_length=6, max_length=15, regex=r"^satori")
):
    """此时的 password 默认值为 'satori'，并且传递的时候必须要以 'satori' 开头
       但是值得注意的是 password 后面的是 str，不再是 Optional[str]，因为默认值不是 None 了
       当然这里即使写成 Optional[str] 也是没有什么影响的
    """
    return {"password": password}


#  ******  post  method *********
class Car(BaseModel):
    name: str  # //必选参数
    weight: Optional[str] = None  # //可选参数
    price: float  # //必选参数
    length: Optional[float] = None


#  ******文件上传 post json ******
@app.post("/car/", tags=["car"])
async def create_item(car: Car):
    return car


# 表单提交 + 文件上传  # 如果要上传多个文件 files: List[bytes] = File(...) or  files: List[UploadFile] = File(...)):
@app.post("/files")
async def create_file(
        file1: bytes = File(...),
        file2: UploadFile = File(...),
        token: str = Form(...),
        username: str = Form(...),  # 表单
        password: str = Form(...)):
    file = file2.file
    return {
        "filesize": len(file1),
        "token": token,
        "oen_content_type": file2.content_type
    }


#   ******  put  method  ******
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None


@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Item, q: Optional[str] = None):
    result = {"item_id": item_id, **item.dict()}
    if q:
        result.update({"q": q})
    return result


if __name__ == "__main__":
    # 启动服务，因为我们这个文件叫做 base_test.py，所以需要启动 base_test.py 里面的 app
    # 第一个参数 "base_test:app" 就表示:当前文件名字 ，然后是 host 和 port 表示监听的 ip 和端口
    uvicorn.run(app, host="0.0.0.0", port=5555)

 

get  post   body  header  ...

class Item(BaseModel):
    id: str
    title: str

@app.post("/body")
async def get_body(item: Item, request: Request):
    myjson = await request.json()  # dict
    res = {
        # 获取 Request Body
        "body": await request.json(),
        "body_bytes": await request.body()
    }
    print(type(myjson))
    return res

 

 统一数据检验

#!/usr/bin/env python3
import time
from typing import Optional
import uvicorn
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from fastapi.responses import JSONResponse
# -------------------------  API  ------------------------------
app = FastAPI(title="Sea test API")
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/airline/test", tags=["Sea"])
def base_test_api():
    return {"你好": "今天天气不错!"}


"""
统一异常处理 （数据校验）
"""
@app.exception_handler(RequestValidationError)
async def request_validation_exception_handler(request: Request, exc: RequestValidationError):
    print(f"参数不对{request.method} {request.url}")
    return JSONResponse({"code": "2003", "message": exc.errors()})


class FlightInfo(BaseModel):
    flight: str  # 必填
    origin: Optional[str] = None  # 可选参数
    destination: str
    etd: str
    eta: str = None


@app.post("/airline/add_flight", tags=["Sea"])
async def add_flight_schedule(flight: FlightInfo, req: Request):
    if flight:
        body = await req.json()
        print(body)
        body["_id"] = time.strftime('%Y-%m-%d') + str(flight.flight)
        body["ts"] = int(time.time())
        body["query_date"] = time.strftime('%Y-%m-%d %T')
    return {"code": "200", "message": "ok"}



if __name__ == '__main__':
    uvicorn.run(app, host='0.0.0.0', port=7500)

 

 

数据嵌套 LIst[Bean]

from typing import Optional, Union, List
import uvicorn
from fastapi import FastAPI, Body
from pydantic import BaseModel, HttpUrl
app = FastAPI()
class Image(BaseModel):
    name: str
    url: HttpUrl
class Person(BaseModel):
    name: str
    desc: Optional[str] = None
    year: List[int]
    img: Union[List[Image], None] = None 
@app.post("/item")
async def create_item(person: Person = Body(embed=True)):
    return person
if __name__ == '__main__':
    uvicorn.run(app=app, host="0.0.0.0", debug=True)

 

 

 

整合eureka (pip install py_eureka_client)

 

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# @mail    : [email protected]
# @Time    : 2022/9/8 15:53
# @Author  : Sea
# @File    : fastapi_euraka.py
# @history:
#     pip install py_eureka_client
# ****************************
import base64
import uvicorn
from fastapi import FastAPI
from fastapi.responses import FileResponse
import py_eureka_client.eureka_client as eureka_client
import socket


# 获取ip
def extract_ip():
    st = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        st.connect(('10.255.255.255', 1))
        ip = st.getsockname()[0]
    except Exception:
        ip = '127.0.0.1'
    finally:
        st.close()
    return ip


server_port = 9090
server_host = extract_ip()
# 配置eureka
eureka_client.init(eureka_server="http://root:root@hadoop002:8761",
                   app_name="fastapi_euraka-app",
                   instance_port=server_port,
                   instance_host=server_host)

app = FastAPI(title="Sea test python euraka")


@app.get("/hello/{name}")
def say_hello(name: str):
    return {"message": f"Hello {name}"}


@app.post("/picture")
def get_picture():
    return FileResponse('/home/sea/img/test.png')


# return ase64
@app.get("/base64img")
def get_picture():
    with open('/home/sea/img/test.png', mode='rb') as file:
        read = file.read()
        image_base64 = base64.standard_b64encode(read)
    return {"image": str(image_base64)}


if __name__ == '__main__':
    eureka_client.get_client()
    uvicorn.run(app, host="0.0.0.0", port=5000)

 

 

 

打开swagger文档：

127.0.0.1:5555/docs

 

 

整合nacos:

import nacos
import uvicorn
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from suds.client import Client
from suds.transport.https import HttpAuthenticated
 
client=nacos.NacosClient('172.16.244.200:8848',namespace='8b6f08b7-b09c-4181-bcac-1505fba5e1aa')
 
async def beat():
    client.add_naming_instance('fastapi-service','172.16.244.10',8000,group_name='dev')
    
 
# 微服务注册nacos
def register_nacos():
    client.add_naming_instance('fastapi-service','172.16.244.10',8000,group_name='dev')
 
 
app=FastAPI()
 
# 微服务注册
register_nacos()
 
 
@app.on_event('startup')
def init_scheduler():
 
    scheduler = AsyncIOScheduler(timezone="Asia/Shanghai")
    scheduler.add_job(beat, 'interval', seconds=5)
    scheduler.start()
 
@app.get('/sap/materials')
async def sap_materials():
   ''' 功能实现'''
  return JSONResponse({'code':1000,'msg':'succ','data':data},status_code=200)

要注意的几点

group_name要与其他java微服务配置要一致 写的不正确 spring cloud gateway 提示404错误

timezone='Asia/Shangai'  不加启动时会提示时区用utc

标签：name,示例,flask,fastapi,app,str,import,euraka,def
From： https://www.cnblogs.com/oceaning/p/17173702.html