Python Fast api advanced setting

Httpx

fast API에서는 결국 최대한 I/O Blocking을 제거하고 asyn I/O를 수행하는 것이 관건

• 기존의 requests 모듈의 request 요청은 I/O blockiing이므로 fastAPI에서 사용하지 않는 것이 좋다.
• requests의 대체제로 **httpx** 모듈을 사용

httpx에서는 Client object를 사용해 requests를 전달하는 방식을 권장

• AsyncClient를 fastAPI lifespan에서 초기화해서 사용

import httpx
from fastapi import FastAPI, Request
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.requests_client = httpx.AsyncClient()
    yield
    await app.requests_client.aclose()

app = FastAPI(lifespan=lifespan)

@app.get("/dogs")
async def get_red_dogs(request: Request):
    requests_client = request.app.requests_client
    response = await requests_client.post("<http://localhost:5000>", data={"hello": "world"})
    return response.json()

• @contextmanager 데코레이터를 제네레이터 함수에 선언하면, 제네레이터 함수는 컨텍스트 매니저로 변경된다. 그리고 아래와 같이 동작한다.
  • with 문으로 들어갈 때 yield까지 실행되고 yield의 값을 반환한다.
  • with 문이 끝날 때 yield 이후의 내용이 실행된다.

Tenacity 라이브러리를 사용하면 AsyncClient에 retry 로직을 적용할 수 있다.

• pip install tenacity

from httpx import ConnectTimeout
from tenacity import retry, stop_after_attempt, retry_if_exception_type

@retry(retry=retry_if_exception_type(ConnectTimeout), stop=stop_after_attempt(3))
async def send_address_match_request(requests_client, payload):
    response = await requests_client.post(
        url=f"{os.getenv('ADDRESS_MATCHING_HOST')}/v4_0/match",
        data=payload,
    )
    response.raise_for_status()
    resp_data = response.json()
    return resp_data

참고

HTTPX

Best way to make Async Requests with FastAPI… the HTTPX Request Client & Tenacity!

Python : async with를 이용한 asyncContextManager 사용

Middleware

middleware : 특정 path의 requests가 수행되기 전이나, response를 반환하기 전에 실행되는 함수

• request 수행 전이나(pre-processing), response 반환 전(post-processing)의 처리 로직

특정 dependency에 대해 yield 명령어가 포함되어 있다면 yield 이후의 코드는 middleware 이후에 수행된다.

background task 또한 middleware 이후에 수행된다.

import time

from fastapi import FastAPI, Request

app = FastAPI()

@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

• request : 입력받는 request
• call_next : request를 파라미터로 받는 path operation
  • middleware 적용을 받는 path operation

Advanced Middleware

이미 fastapi 내부에서 구현된 middleware가 다수 존재

• HTTPSRedicrectMiddleware : http나 ws로 들어온 모든 request를 https로 리다이렉트
• TrustedHostMiddleware : 모든 request가 Host headers가 적절히 설정되었는지 확인
  • HTTP Host Header Attacks을 방어

  from fastapi import FastAPI
  from fastapi.middleware.trustedhost import TrustedHostMiddleware
  
  app = FastAPI()
  
  app.add_middleware(
      TrustedHostMiddleware, allowed_hosts=["example.com", "*.example.com"]
  )
  
  @app.get("/")
  async def main():
      return {"message": "Hello World"}
  

• GZipMiddleware : requests에 Accept-Encoding에 gzip이 설정되어 있는 경우에 response를 gzip으로 압축
• from fastapi import FastAPI from fastapi.middleware.gzip import GZipMiddleware app = FastAPI() app.add_middleware(GZipMiddleware, minimum_size=1000) @app.get("/") async def main(): return "somebigcontent"

참고

Advanced Middleware - FastAPI

lifespan

lifespan : 서버 어플리케이션이 start up 되기 전이나 shutdown 되기 전에 로직을 넣을 수 있다.

• 모든 requests에서 전역적으로 사용해야 하는 설정이나 리소스 등을 세팅할 때 주로 사용
  • ex) db connection pool ..
  • ex) ml model initializaion

from contextlib import asynccontextmanager

from fastapi import FastAPI

def fake_answer_to_everything_ml_model(x: float):
    return x * 42

ml_models = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Load the ML model
    ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model
    yield
    # Clean up the ML models and release the resources
    ml_models.clear()

app = FastAPI(lifespan=lifespan)

@app.get("/predict")
async def predict(x: float):
    result = ml_models["answer_to_everything"](x)
    return {"result": result}

• lifespan에서 yield 이전까지는 application이 start-up될 때 수행
• lifespan에서 yield 이후는 application이 shut-down될 때 수행

asynccontextmanager

• 기존의 with 구문을 async로 처리

async with lifespan(app):
    await do_stuff()

lifespan은 sub mount application에서는 수행되지 않는다.

• 오직 main application에서만 실행됨