閱讀目錄

一、部署之殤

二、日志之殤

三、中間件之殤

四、配置文件之殤

五、其它

一、部署之殤

1 linux后臺啟動

nohup uvicorn main:app --host 0.0.0.0 --port 8080

2 Docker部署

FROM python:3.7

RUN pip install fastapi uvicorn

EXPOSE 80

COPY ./app /app

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080"]

2.1 Docker + gunicorn

gunicorn配置文件

#!usr/bin/env python

# encoding: utf-8

import multiprocessing

# 監(jiān)聽端口

bind = '0.0.0.0:8899'

# 工作模式

worker_class = 'uvicorn.workers.UvicornWorker'

# 并行工作進程數(shù)

workers = multiprocessing.cpu_count() * 2 + 1

# 設(shè)置守護進程

#daemon = True

# 配置文件方式配置日志

logconfig = "./logger.ini"

Dockerfile

FROM python:3.7

ENV TZ Asia/Shanghai

#將項目代碼放入鏡像

COPY . /app

WORKDIR /app

#安裝第三方模塊，更新數(shù)據(jù)庫

RUN pip install -r requirements.txt -i http://mirrors.aliyun.com/pypi/simple --trusted-host mirrors.aliyun.com \

&& rm -rf configure

ENTRYPOINT ["gunicorn", "-c", "gunicorn.conf.py", "main:app"]

3 k8s部署

3.1 service.yaml

apiVersion: v1

kind: Service

metadata:

name: project_name # 項目名稱

spec:

ports:

- name: http

port: 80

targetPort: 8899

type: ClusterIP

3.2 deployment.yaml

apiVersion: apps/v1

kind: Deployment

metadata:

name: project_name # 項目名稱

spec:

template:

spec:

imagePullSecrets:

- name: registry-pull-secret

containers:

- name: project_name # 項目名稱

image: registry-vpc.cn-shanghai.aliyuncs.com/xxx/project_name:lates # 鏡像

imagePullPolicy: Always

volumeMounts:

- name: host-time

mountPath: /etc/localtime

volumes:

- hostPath:

path: /etc/localtime

name: host-time

二、日志之殤

1 日志配置

日志配置文件，本地環(huán)境、測試環(huán)境、生產(chǎn)環(huán)境可以配置不同的日志的打印

[loggers]

;這里面把uvicorn創(chuàng)建的logger配置都覆蓋了，注意最后一個`,`不能缺少、防止日志多次打印

keys=root, gunicorn.error, gunicorn.access,uvicorn.error,uvicorn.access,

[handlers]

keys=error_file, access_file

[formatters]

keys=generic, access

[logger_root]

level=DEBUG

handlers=access_file

[logger_]

level=INFO

handlers=access_file

qualname=

propagate=0

[logger_uvicorn.error]

level=INFO

handlers=error_file

qualname=uvicorn.error

propagate=0

[logger_uvicorn.access]

level=INFO

handlers=access_file

qualname=uvicorn.access

propagate=0

[logger_gunicorn.error]

level=INFO

handlers=error_file

propagate=1

qualname=gunicorn.error

[logger_gunicorn.access]

level=INFO

handlers=access_file

propagate=0

qualname=gunicorn.access

;注意日志配置的地址

[handler_error_file]

class=logging.FileHandler

formatter=generic

args=('/app/log/error.log',)

[handler_access_file]

class=logging.FileHandler

formatter=access

args=('/app/log/access.log',)

[formatter_generic]

format=[%(asctime)s] %(levelname)s in %(module)s: %(message)s

datefmt=%Y-%m-%d %H:%M:%S

class=logging.Formatter

;配置日志打印的信息

[formatter_access]

format=[%(asctime)s] %(levelname)s in %(module)s: %(message)s

class=logging.Formatter

2 讀取配置

方案：讀取文件 or 啟動時設(shè)配置

# 環(huán)境變量

fast_api_env = os.environ.get('FAST_API_ENV')

# 獲取logger對象

def get_logger(filename="logger.ini", logger_name='root'):

logging.config.fileConfig(fname=filename, disable_existing_loggers=False)

return logging.getLogger(logger_name)

def init_log():

"""初始化日志"""

print("加載log文件...")

try:

global common_config

if fast_api_env == 'local':

# 本地環(huán)境

LOG_CONFIG_PATH = os.path.join(BASE_DIR, 'conf', 'logger-local.ini')

# logger = get_logger(os.path.join(BASE_DIR, 'conf', 'logger.ini'))

else:

common_config.LOG_CONFIG_PATH = os.path.join(BASE_DIR, 'conf', 'logger-prod.ini')

# logger = get_logger(common_config.LOG_CONFIG_PATH, logger_name='file')

except Exception as e:

raise LogConfigError(e)

3 啟動配置logger.ini

uvicorn.run(app, host='0.0.0.0', port=8899, log_config=common_config.LOG_CONFIG_PATH)

*配置完成后，logging.debug()等使用即可

三、中間件之殤(自定義中間件)

1 @app.middleware("http")

@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

# 添加響應(yīng)頭

response.headers["X-Process-Time"] = str(process_time)

return response

2 app.add_middleware

from starlette.datastructures import Headers

from starlette.responses import PlainTextResponse

from starlette.types import ASGIApp, Receive, Scope, Send

class AuthMiddleware:

def __init__(self, app: ASGIApp) -> None:

self.app = app

async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:

logging.info(scope.get('path'))

if scope.get('path'):

url = URL(scope=scope)

if url.path not in common_config.WHITE_LIST: # 設(shè)置白名單

headers = Headers(scope=scope)

token = headers.get("Token")

# 自定義訪問攔截

if not token or not headers.get("username") or not check_devops_auth(token): # 自定義驗證token，和其他請求信息作為認證攔截

response = PlainTextResponse("未登陸用戶", status_code=401)

await response(scope, receive, send)

return

await self.app(scope, receive, send)

app.add_middleware(AuthMiddleware)

四、配置文件之殤

方案：采用ini配置文件，讀取后寫入全局變量

1 配置文件

[common]

PROJECT_NAME = kk-jira-monitor

BACKEND_CORS_ORIGINS = http://127.0.0.1:8080,

API_V1_STR = /api/v1.0

[mysql]

USERNAME = admin

PASSWORD = 123456

HOST =

PORT = 3306

DATABASE =

SQLALCHEMY_DATABASE_URI = mysql+pymysql://%(USERNAME)s:%(PASSWORD)s@%(HOST)s:%(PORT)s/%(DATABASE)s

2 初始化

common_config = None

mysql_config = None

def init_config():

"""初始化配置文件"""

global mysql_config, common_config

print("加載配置文件...")

config = ReConfigParser()

try:

if fast_api_env == 'local':

config.read(os.path.join(BASE_DIR, 'conf', 'conf-local.ini'), encoding='utf-8')

elif fast_api_env == 'dev':

config.read(os.path.join(BASE_DIR, 'conf', 'conf-dev.ini'), encoding='utf-8')

else:

config.read(os.path.join(BASE_DIR, 'conf', 'conf-prod.ini'), encoding='utf-8')

mysql_config = MySQLConfig(**dict(config.items('mysql')))

common_config = CommonConfig(**dict(config.items('common')))

except Exception as e:

# logger.exception(f"配置文件初始化失敗,{e.__str__()}")

raise ConfigError(e)

3 configparse

from configparser import ConfigParser

class ReConfigParser(ConfigParser):

def __init__(self, defaults=None):

ConfigParser.__init__(self, defaults=defaults)

"""復(fù)寫方法實現(xiàn)key值區(qū)分大小寫"""

def optionxform(self, optionstr):

return optionstr

4 配置變量驗證

import os

from typing import Optional

from pydantic import BaseModel

class CommonConfig(BaseModel):

SECRET_KEY: str = os.urandom(32)

PROJECT_NAME: str

API_V1_STR: str

# 允許訪問的origins

BACKEND_CORS_ORIGINS: str

class MySQLConfig(BaseModel):

USERNAME: str = None

PASSWORD: str = None

HOST: str = None

PORT: int = None

DATABASE: str = None

SQLALCHEMY_DATABASE_URI: str = (

f"mysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}"

)

五、其它

1 問題一(中間件執(zhí)行報錯)

ASGI 'lifespan' protocol appears unsupported.

@app.on_event('startup') 將不會執(zhí)行

2 問題二(定時任務(wù)報錯)

借助的apshechduler注冊的定時任務(wù)如果執(zhí)行報錯，捕獲不到異常信息

解決辦法可見 分離定時任務(wù)

總結(jié)

以上是生活随笔為你收集整理的fastapi日志重复打印_【FastAPI】踩坑总结的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。