[udemy] Complete FastAPI masterclass from scratch - 학습 정리 2

1. middleware

   1) @app.middleware 데코레이터 사용

      - FastAPI의 내장 데코레이터로, 간단한 미들웨어를 정의할 때 적합하다.
      - 동기적 또는 비동기적 함수로 작성할 수 있다.
      - 요청(Request)과 응답(Response)을 가로채서 처리한다.

from fastapi import FastAPI, Request
from fastapi.responses import Response

app = FastAPI()

@app.middleware("http")
async def simple_middleware(request: Request, call_next):
    # 요청 전 처리
    print("Before Request")
    response = await call_next(request)
    # 응답 후 처리
    print("After Request")
    return response

 

   2) BaseHTTPMiddleware 상속

      - starlette.middleware.base.BaseHTTPMiddleware를 상속받아 사용하며, 더 복잡하고 구조적인 미들웨어를 작성할 때 적합하다.
      - 요청/응답 처리에 상태 관리나 특정 로직을 삽입할 수 있다.
      - dispatch 메서드를 오버라이드하여 요청 처리 로직을 커스터마이즈한다.

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()

class CustomMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 요청 전 처리
        print("Custom Middleware: Before Request")
        response = await call_next(request)
        # 응답 후 처리
        print("Custom Middleware: After Request")
        return response

app.add_middleware(CustomMiddleware)

 

   3) 주요 차이점

특성	@app.middleware	BaseHTTPMiddleware
구현 난이도	쉬움	조금 복잡
확장성	제한적	구조적이고 확장 가능
사용 사례	간단한 요청/응답 전후 작업	상태 관리, 로깅, 고급 로직 구현
추가 상태 관리	불편함	용이
Starlette 의존성	없음	필요

 

   4) BaseHTTPMiddleware에서의 확장성

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse

app = FastAPI()

class BaseCustomMiddleware(BaseHTTPMiddleware):
    def __init__(self, app):
        super().__init__(app)
        self.state = {"request_count": 0}

    async def before_request(self, request: Request):
        """요청 전 처리 작업"""
        self.state["request_count"] += 1
        print(f"Request #{self.state['request_count']}")

    async def after_request(self, response):
        """응답 후 처리 작업"""
        response.headers["X-Custom-Header"] = "Middleware Active"
        print("Response modified with custom header")
        return response

    async def dispatch(self, request: Request, call_next):
        # 요청 전 작업 실행
        await self.before_request(request)

        try:
            response = await call_next(request)
        except Exception as e:
            # 예외 처리
            return JSONResponse({"error": str(e)}, status_code=500)

        # 요청 후 작업 실행
        response = await self.after_request(response)

        return response

# 미들웨어 상속
class ExtendedMiddleware(BaseCustomMiddleware):
    async def before_request(self, request: Request):
        """상속받은 미들웨어의 요청 전 처리 로직 확장"""
        await super().before_request(request)
        print("Extended Middleware Before Request Logic")

app.add_middleware(ExtendedMiddleware)

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

 

2. logging

   1) 의존성 설치 

pip install fastapi uvicorn pyyaml

 

   2) 디렉토리 구조

.
├── main.py
├── module_a.py
├── module_b.py
├── logging_config.yaml
└── logs/
    ├── app.log
    └── download_log.txt

 

   3) logging_config.yaml - 로깅 설정 파일

      - 별도의 app_logger를 생성하지 않아도 root 로거를 통해 로깅이 처리된다.

version: 1
disable_existing_loggers: false
formatters:
  detailed:
    format: "%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s"

handlers:
  console:
    class: logging.StreamHandler
    formatter: detailed
    level: DEBUG

  file:
    class: logging.FileHandler
    formatter: detailed
    level: DEBUG
    filename: logs/app.log

  download_file:
    class: logging.FileHandler
    formatter: detailed
    level: INFO
    filename: logs/download_log.txt

loggers:
  app:
    level: DEBUG
    handlers: [console, file]
    propagate: false

  download:
    level: INFO
    handlers: [download_file]
    propagate: false

root:
  level: DEBUG
  handlers: [console, file]

 

   4) main.py - FastAPI 메인 파일

import logging.config
import yaml
from fastapi import FastAPI, Request
from module_a import module_a_function
from module_b import module_b_function

# 로깅 설정 로드
with open("logging_config.yaml", "r") as file:
    config = yaml.safe_load(file)
    logging.config.dictConfig(config)

# FastAPI 앱 생성
app = FastAPI()

# 로거 정의
app_logger = logging.getLogger("app")
download_logger = logging.getLogger("download")

@app.get("/")
async def read_root():
    app_logger.info("Root endpoint accessed.")
    return {"message": "Welcome to the FastAPI app!"}

@app.get("/module-a")
async def call_module_a():
    result = module_a_function()
    app_logger.debug("Module A function called.")
    return {"result": result}

@app.get("/module-b")
async def call_module_b():
    result = module_b_function()
    app_logger.debug("Module B function called.")
    return {"result": result}

@app.post("/download")
async def download_endpoint(request: Request):
    user_id = request.headers.get("X-User-ID", "unknown")
    download_logger.info(f"User {user_id} initiated a download.")
    app_logger.info(f"Download request received from user {user_id}.")
    return {"message": "Download started."}

 

   5) module_a.py - 모듈 A

import logging

logger = logging.getLogger("app")

def module_a_function():
    logger.info("Executing Module A function.")
    return "Result from Module A"

 

   6) module_b.py - 모듈 B

import logging

logger = logging.getLogger("app")

def module_b_function():
    logger.info("Executing Module B function.")
    return "Result from Module B"

 

   7) 로그 포맷:

      - %(asctime)s : 로그 발생 시간
      - %(name)s : 로거 이름
      - %(levelname)s : 로그 레벨
      - %(filename)s:%(lineno)d : 파일명 및 코드 줄 번호

 

   8) 다중 핸들러:

      - console : 콘솔 출력
      - file : app.log 파일 출력
      - download_file : download_log.txt 파일 출력

 

   9) 로깅 레벨:

      - DEBUG : 디버그 로그까지 출력
      - INFO : 정보 로그까지만 출력

 

3. WebSocket

   1) FastAPI WebSocket 기본 사용 방법

      - FastAPI와 WebSocket 클래스 임포트
      - app.websocket 데코레이터를 사용하여 WebSocket 엔드포인트 정의
      - 클라이언트 연결 요청 처리 (WebSocket.accept())
      - 메시지 송수신 처리 (WebSocket.receive_text(), WebSocket.send_text())

from fastapi import FastAPI, WebSocket

app = FastAPI()

clients = []

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()  # 클라이언트 연결 수락
    while True:
        data = await websocket.receive_text()  # 클라이언트로부터 메시지 수신
        for client in clients:
        	# await websocket.send_text(f"Message received: {data}")  # 클라이언트로 메시지 송신
            await client.send_text(data)

 

   2) Broadcast 기능

      - FastAPI로 다중 클라이언트와 통신할 때, 연결된 모든 클라이언트에게 메시지를 브로드캐스트하는 기능이 필요할 수 있다. 이를 위해 set을 사용하여 활성 WebSocket 연결을 관리한다.

from fastapi import FastAPI, WebSocket
from typing import List

app = FastAPI()

connected_clients: List[WebSocket] = []

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    connected_clients.append(websocket)  # 연결된 클라이언트를 목록에 추가
    try:
        while True:
            data = await websocket.receive_text()
            for client in connected_clients:
                await client.send_text(f"Broadcast: {data}")
    except:
        connected_clients.remove(websocket)  # 연결 해제 시 클라이언트 제거

 

   3) SSE(Server-Sent Events) 구현

      - StreamingResponse를 사용하여 스트림 데이터를 지속적으로 전송

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def event_stream():
    while True:
        # 예시로 서버에서 현재 시간을 매초 보내는 이벤트 스트림 생성
        yield f"data: The time is {str(datetime.datetime.now())}\n\n"
        await asyncio.sleep(1)

@app.get("/sse")
async def sse_endpoint():
    return StreamingResponse(event_stream(), media_type="text/event-stream")

 

      - 클라이언트는 EventSource API를 사용해 SSE 데이터를 받을 수 있음

const eventSource = new EventSource("/sse");
eventSource.onmessage = (event) => {
    console.log(event.data);
};

 

   4) 간단한 채팅 앱 구현

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/chat")
async def chat_websocket(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client says: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast("A client disconnected.")

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Chat</title>
</head>
<body>
    <h1>WebSocket Chat</h1>
    <div id="chat-box" style="border: 1px solid black; height: 300px; overflow-y: scroll; margin-bottom: 10px;">
    </div>
    <input id="message" type="text" placeholder="Type your message" />
    <button onclick="sendMessage()">Send</button>

    <script>
        const chatBox = document.getElementById("chat-box");
        const messageInput = document.getElementById("message");
        const ws = new WebSocket("ws://localhost:8000/chat");

        ws.onmessage = (event) => {
            const message = document.createElement("div");
            message.textContent = event.data;
            chatBox.appendChild(message);
            chatBox.scrollTop = chatBox.scrollHeight; // Scroll to bottom
        };

        function sendMessage() {
            const message = messageInput.value;
            ws.send(message);
            messageInput.value = "";
        }
    </script>
</body>
</html>

 

 

4. Dependencies

   1) Class dependencies

      - 엔드포인트 뷰의 입력 인자로 정의된 변수들은 depends()의 함수에 정의된 입력 인자와 이름이 동일하다면, mapping되어 동일한 값의 변수를 사용할 수 있다.

class Account:
  def __init__(self, name: str, email: str):
    self.name = name
    self.email = email

@router.post('/user')
def create_user(name: str, email: str, password: str, account: Account = Depends()): 
# account: Account = Depends(Account)은 account: Account = Depends()와 동일하다.
  # account - perform whatever operations
  return {
    'name': account.name,
    'email': account.email
  }

 

   2) Multi level dependencies

def convert_params(request: Request, separator: str):
  query = []
  for key, value in request.query_params.items():
    query.append(f"{key} {separator} {value}")
  return query

def convert_headers(request: Request, separator: str = '--', query = Depends(convert_params)):
  out_headers = []
  for key, value in request.headers.items():
    out_headers.append(f"{key} {separator} {value}")
  return {
    'headers': out_headers,
    'query': query
  }

@router.get('')
def get_items(test: str, separator: str = '--', headers = Depends(convert_headers)):
  return {
    'items': ['a', 'b', 'c'],
    'headers': headers
  }

 

   3) Global dependencies

from fastapi import FastAPI, APIRouter, Depends, Request
import logging

# 로거 설정
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(message)s")
logger = logging.getLogger(__name__)

# log 함수 정의
async def log(request: Request):
    logger.info(f"Request URL: {request.url}")

# 라우터 생성
router = APIRouter(
    prefix="/dependencies",
    tags=["dependencies"],
    dependencies=[Depends(log)]
)

# 엔드포인트 정의
@router.get("/example")
async def example_endpoint():
    return {"message": "This is an example endpoint"}

@router.get("/another")
async def another_endpoint():
    return {"message": "This is another endpoint"}

# FastAPI 애플리케이션 생성 및 라우터 등록
app = FastAPI()
app.include_router(router)

 

5. OCR application

   1) 라이브러리 설치

      1. Tesseract OCR :

         - Tesseract는 Google에서 유지 관리하는 오픈 소스 OCR(Optical Character Recognition) 엔진이다. 이미지에서 텍스트를 추출하는 데 사용된다.

 

         - 특징 :
            - 다국어 지원: 100개 이상의 언어를 지원하며, 언어 데이터를 추가로 설치 가능
            - 다양한 출력 형식: 텍스트, HOCR, TSV, PDF 등으로 결과를 저장 가능
            - 확장성: 사용자 정의 언어 데이터 학습 가능
            - 호환성: 다양한 플랫폼(Linux, Windows, macOS)에서 작동하며 Python과 같은 언어에서 사용 가능(pytesseract)

sudo apt update
sudo apt install tesseract-ocr
sudo apt install libtesseract-dev

 

         - 추가 언어 데이터 설치 :

            - 한국어 : tesseract-ocr-kor
            - 중국어(간체) : tesseract-ocr-chi-sim
            - 아랍어 : tesseract-ocr-ara

sudo apt install tesseract-ocr-langcode

 

      2. Pytesseract :

         - pytesseract는 Tesseract-OCR 엔진을 Python에서 사용할 수 있도록 해주는 래퍼이다. 이미지를 입력으로 받아 텍스트를 추출하는 데 사용된다.

 

         - 특징 :
            - 이미지에서 텍스트를 추출 (Optical Character Recognition, OCR)
            - 다국어 지원 (추가 언어 패키지 설치 필요)
            - 이미지 파일, NumPy 배열 등 다양한 입력 형식 지원

sudo apt-get install tesseract-ocr

 

         - 주요 메서드 :
            - pytesseract.image_to_string(image, lang='eng') : 이미지를 문자열로 변환
            - pytesseract.image_to_boxes(image) : 이미지에서 문자 박스 정보 반환
            - pytesseract.image_to_data(image) : 이미지에서 텍스트와 위치 정보를 포함한 데이터 반환
            - pytesseract.get_languages(config='') : 설치된 언어 확인

 

         - 샘플 코드 : 

from fastapi import FastAPI, File, UploadFile
import shutil
import pytesseract

app = FastAPI()

@app.post('/ocr')
def ocr(image: UploadFile = File(...)):
  filePath = 'txtFile'
  with open(filePath, "w+b") as buffer:
    shutil.copyfileobj(image.file, buffer)
  return pytesseract.image_to_string(filePath, lang='eng')

 

6. Blog site - FastAPI

   1) 저장소 및 내용 정리

      - 저장소 : https://github.com/CatalinStefan/fastapi-blog-api

      - 내용 정리 :

         - 기본적인 내용으로 따로 정리할 사항이 없음

 

7. Blog site - FastAPI

   1) 저장소 및 내용 정리

      - 저장소 : https://github.com/CatalinStefan/instagram-clone-api

      - 내용 정리 :

         - 기본적인 내용으로 따로 정리할 사항이 없음

 

8. Warehouse app with Microservices and Redis

   1) 저장소 및 내용 정리

      - 저장소 : https://github.com/CatalinStefan/fastapi-microservices

      - 내용 정리 :

         - 라이브러리 및 코드의 설명이 부족하여 별도로 추가 정리함