轻量快速的 Python ASGI uvicorn 框架使用
Published in：2023-05-15 |
Words: 1.2k | Reading time: 6min | reading：

轻量快速的 Python ASGI uvicorn 框架使用

下载使用

1
2
pip install fastapi
pip install uvicorn

优点

Python 仍缺乏异步的网关协议接口，ASGI 的出现填补了这一空白，现在开始，我们能够使用共同的标准为所有的异步框架来实现一些工具，ASGI 帮助 Python 在 Web 框架上和 Node.JS 及 Golang 相竟争，目标是获得高性能的 IO 密集型任务，ASGI 支持 HTTP2 和 WebSockets，WSGI 是不支持的。

Uvicorn 目前支持 HTTP1.1 和 WebSocket，计划支持 HTTP2

框架使用

启动

  • 1.脚本式
1
uvicorn example:app
  • 2.代码式
1
2
3
4
5
6
7
8
import uvicorn

async def app(scope, receive, send):
    ...

if __name__ == "__main__":
    uvicorn.run("example:app", host="127.0.0.1", port=5000, log_level="info")

  • 3.整合 FastAPI
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

# 路由注解 参数注解
@router.get("/log/log", summary='summary', description='desc')
def read_item(item_id: Union[str, None] = Query(default=None, alias="log_id")):
    """
    获取报告生成进度
    @:parameter item_id log id
    """
    return null;

if __name__ == '__main__':
    uvicorn.run(app=app)

功能引入

  • 1.全局 log

启动文件配置

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

def init_app():
    app = FastAPI(title="", version="0.0.1",
                  description="", debug=True)
    app.include_router(router=api_router, prefix="/api/v1")
    logging.getLogger().handlers = [InterceptHandler()]
    logger.configure(
        handlers=[{"sink": sys.stdout, "level": logging.DEBUG, "format": format_record}])
    logger.add(LOG_DIR, encoding='utf-8', rotation="9:46")
    logger.debug('日志系统已加载')
    logging.getLogger("uvicorn.access").handlers = [InterceptHandler()]
    return app


app = init_app()

log 基础配置

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
import logging
import os
from pprint import pformat

from loguru import logger

from http_server.log.log_base import LOG_FORMAT
import logging
from loguru import logger
# from ..config import config
from loguru._defaults import LOGURU_FORMAT


class InterceptHandler(logging.Handler):
    def emit(self, record):
        # Get corresponding Loguru level if it exists
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        # Find caller from where originated the logged message
        frame, depth = logging.currentframe(), 2
        while frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1

        logger.opt(depth=depth, exception=record.exc_info).log(
            level, record.getMessage()
        )


def format_record(record: dict) -> str:
    format_string = LOG_FORMAT

    if record["extra"].get("payload") is not None:
        record["extra"]["payload"] = pformat(
            record["extra"]["payload"], indent=4, compact=True, width=88
        )
        format_string += "\n<level>{extra[payload]}</level>"

    format_string += "{exception}\n"
    return format_string

# 基础变量配置
import os
import time

# -----------------------系统调试------------------------------------
DEBUG = True
# -----------------------日志-----------------------------------------
LOG_DIR = os.path.join(os.getcwd(), f'log_dir\\{time.strftime("%Y-%m-%d")}.log')
LOG_FORMAT = '<level>{level: <8}</level>  <green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> - <cyan>{name}</cyan>:<cyan>{' \
             'function}</cyan> - <level>{message}</level> '

  • 2.全局异常处理

异常类定义

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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
# from rest_framework.exceptions import APIException
#
#
# class Success(APIException):
#     code = 201
#     msg = 'ok'
#     error_code = 0
#
#
# class DeleteSuccess(APIException):
#     code = 202
#     msg = 'delete ok'
#     error_code = 1
#
#
# class UpdateSuccess(APIException):
#     code = 200
#     msg = 'update ok'
#     error_code = 2
#
#
# class ServerError(APIException):
#     code = 500
#     msg = 'sorry, we made a mistake!'
#     error_code = 999
#
#
# class ParameterException(APIException):
#     code = 400
#     msg = 'invalid parameter'
#     error_code = 1000
#
#
# class NotFound(APIException):
#     code = 404
#     msg = 'the resource are not found'
#     error_code = 1001
#
#
# class AuthFailed(APIException):
#     code = 401
#     msg = 'authorization failed'
#     error_code = 1005
#
#
# class Forbidden(APIException):
#     code = 403
#     error_code = 1004
#     msg = 'forbidden, not in scope'
from docutils.nodes import status
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from requests import Request
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
from http_server.router import router

"""
自定义异常
"""


class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name


@router.api_router.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),
    )


@router.api_router.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)


@router.api_router.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )

使用

1
2
3
4
5
6
7
8
9
10
11
12
@router.get("/{id}", summary='测试接口', description='测试接口')
def read_root(id: int = Path(title="id not null")):
    """
    测试接口
    """
    if id == 1:
        raise HTTPException(404, "no data")
    datas = {
        "hello ": "wprils"
    }
    return JsonResponse.success(datas)

  • 3.路由定义

启动文件配置

1
app.include_router(router=api_router, prefix="/api/v1")

路由文件配置

1
2
3
4
5
6
7
8
from fastapi import APIRouter

from http_server.response.JsonResponse import JsonResponse
from http_server.view import log_process

api_router = APIRouter()
api_router.include_router(log_process.router, prefix="/shell", tags=['脚本操作'])

接口定义

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from fastapi import APIRouter

from http_server.response.JsonResponse import JsonResponse
from http_server.request.http_request import data_process, get_csv_data
from http_server.tools.file_process import download_csv_data
from run.application_main import run_process_thread

router = APIRouter()


@router.get("")
def read_root():
    datas = {
        "hello ": "wprils"
    }
    return JsonResponse.success(datas)

  • 4.统一返回值

定义

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
# -*- coding: utf-8 -*-


class JsonResponse(object):
    """
    统一的json返回格式
    """

    def __init__(self, data, code, msg):
        self.data = data
        self.code = code
        self.msg = msg

    @classmethod
    def success(cls, data=None, code=0, msg='success'):
        return cls(data, code, msg)

    @classmethod
    def error(cls, data=None, code=-1, msg='error'):
        return cls(data, code, msg)

    def to_dict(self):
        return {
            "code": self.code,
            "msg": self.msg,
            "data": self.data
        }

使用

1
2
3
4
5
6
@router.get("")
def read_root():
    datas = {
        "hello ": "wprils"
    }
    return JsonResponse.success(datas)

参考链接

python
Prev:
Swagger + Zuul 整合微服务接口文档
Next:
使用 Electron 框架中 http 请求与 Java 后端 websocket 服务端通信

    • Copyright © 2023 - Now Zhaoqi.Cao | Powered by Hexo | Theme Bamboo

    Total visits： times |