Fastapi

Fastapi - interview questions. interviewprep.org Q11~15

monsangter 2025. 1. 10. 03:00


11. FastAPI 는 직렬화를 어떻게 다루고, 데이터를 어떻게 검증하는가.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# 요청 및 응답 모델 정의
class Item(BaseModel):
    name: str
    price: float
    in_stock: bool

class ResponseModel(BaseModel):
    message: str
    item_name: str
    item_price: float
    in_stock: bool

@app.post("/items/", response_model=ResponseModel)
async def create_item(item: Item):
    """
    요청 데이터 검증 및 응답 직렬화
    """
    return {
        "message": "아이템 등록 성공!",
        "item_name": item.name,
        "item_price": item.price,
        "in_stock": item.in_stock
    }

 

FastAPI는 Pydantic 모델을 사용해 데이터의 직렬화와 검증을 수행한다.

Pydantic 모델은 들어오는 요청 데이터와 나가는 응답 데이터 구조를 정의해, 해당 스키마에 맞는지 검증한다.

요청 데이터가 유효하면 파이썬 객체로 변환하고, 응답 데이터는 다시 JSON 형식으로 직렬화 하여 변환한다.

 

 

 

클라이언트가 JSON 데이터를 보낼 때, FastAPI는 Pydantic 모델을 기반으로 데이터가 올바른 타입과 구조를 갖췄는지 검증한다.

스키마에 맞으면 역직렬화하여 파이썬 객체로 변환해 함수에 전달한다.

스키마와 맞지 않는다면 자동으로 400 Bad Request 에러 응답 보낸다.

 

2. FastAPI 함수가 반환하는 Python 객체는 JSON 형식으로 자동 변환된다.

 

 

12. FastAPI 앱에서는 어떻게 단위 테스트를 세팅할 것인가 ? 

먼저 pytest나 httpx, 리퀘스트 등의 라이브러리 설치가 필요하다.

 

이어 test_*.py형식으로 파일을 생성한다.

이 파일에서 테스트할 앱을 비롯해 테스트 클라이언트와 pytest 등의모듈등을 임포트 한다. 

 

from fastapi.testclient import TestClient
from main import app  # FastAPI 애플리케이션 import

client = TestClient(app)  # TestClient를 통해 HTTP 요청 시뮬레이션

 

테스트 클라이언트에 fastapi app 객체 전달.

테스트 클라이언트 객체를 사용해 HTTP 리퀘스트를 시뮬레이트 할 수 있다.

 

def test_read_main():
    response = client.get("/")  # 루트 엔드포인트에 GET 요청
    assert response.status_code == 200  # 응답 코드가 200인지 확인
    assert response.json() == {"message": "Hello World"}  # 응답 본문 확인
    
    
def test_create_item():
    data = {"name": "Laptop", "price": 1500.0, "in_stock": True}
    response = client.post("/items/", json=data)  # POST 요청으로 데이터 전송
    assert response.status_code == 201  # 201 Created
    assert response.json()["name"] == "Laptop"  # 응답 데이터 검증

 

이어 각 엔드포인트에 대해 테스트 함수를 작성한다.

상태코드, 헤더, 바디등의 검증

 

 

테스트 클라이언트 객체를 사용해 HTTP 리퀘스트를 시뮬레이트 할 수 있다.

 

 

외부 api 호출이나 데이터 베이스 연동 테스트를 위해

외부 종속성 Mocking 이 필요할 수도 있다. (실제 호출이나 연동이 아닌 가짜 응답 데이터 사용)

 

Pytext 는 단위 테스트에 필요한 모든 셋업과 tear down 을 지원한다.

 

Fastapi에서 어떻게 백그라운드 작업을 어떻게 구현할 것인가?

 

Fastapi 에서는 BackgroundTasks 클래스를 사용해 응답을 클라이언트에 보낸 후 별도의 작업을 비동기로 처리한다. 이 기능은 로그 작성, 이메일 발송등 시간이 오래 걸리는 작업을 백그라운드에서 처리함으로써 응답 시간을 단축시킨다. 

 

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

# 백그라운드 작업 함수
def write_log(message: str):
    with open("log.txt", "w") as file:
        file.write(message)

@app.post("/send/{message}")
async def send_message(message: str, background_tasks: BackgroundTasks):
    # write_log 함수를 백그라운드 작업으로 등록
    background_tasks.add_task(write_log, message)
    return {"Message": "메시지가 전송되었습니다!"}  # 클라이언트에 즉시 응답

 

BackgroundTasks를 import 하고 백그라운드에서 실행할 함수를 정의한다. write_log 작성

엔드포인트에 인자로 BackgroundTasks 추가한다.

add_task 메서드를 사용해 작업 등록한다.

 

해당 작업은 write_log 함수를 백그라운드 작업으로 등록하여 message를 인자로 전달한다.

이 작업은 클라이언트에 응답을 반환한 후 백그라운드에서 실행한다.

 

로그저장, 이메일발송, 알림전송, 외부 API 호출. 등에 사용한다.

 

코루틴과 같은 비동기 작업과 백그라운드 작업에는 차이가 있는데,

코루틴은 비동기작업을 수행하기 위해 반드시 코루틴을 지원하는 라이브러리를 사용해야한다. 즉 await 키워드를 통해 다른 코루틴을 호출하는 방식으로 동작하기 떄문에 전통적인 동기 라이브러리와는 호환되지 않는 경우가 많다.

 

반면 BackgroundTasks 는 FastAPI 에서 제공하는 비동기 작업 도구로, 동기와 비동기 함수 두개 모두 지원한다. 비동기 이벤트 루프에 함수를 등록해 응답 후 실행한다.

 

비동기 이벤트 루프는 작업큐를 가지고 있어, 해당 작업을 큐에 등록하고, 작업이 완료되면 그 결과를 처리한다. 즉 비동기 작업이 완료되기 전에는 다른 작업을 처리한다.

 

코루틴에서는 작업 완료시 응답을 반환하고, BackgroundTasks에서는 먼저 응답을하고 작업을 처리한다는 차이도 있다.

 

14. HTTP 프로토콜을 의존성 주입이아니라 바로 사용하는 케이스를 보여줘라.

 

FastAPI의 의존성 주입 시스템은, 요청 파싱, 인증, 데이터 검증을 자동으로 처리해 개발을 단순화하여 코드 반복을 줄여준다.

그러나 경우에 따라 HTTP 프로토콜을 직접 다루는 것이 더 적합할 떄도 있다. 특히 저수준 네트워크 작업이나 커스텀 프로토콜 처리가 필요할 떄는 FastAPI의 추상화된 레이어를 거치지 않는 것이 유리하다.

 

WebSocket과 커스텀 프로토콜을 처리하는 경우를 예로 들 수 있다.

웹소켓 서버를 구축하며 바이너리 데이터 스트림이나 특수 인코딩/디코딩 방식을 사용할 경우, FastAPI의 의존성 주입 시스템이 제공하는 기능만으로는 충분하지 않을 수 있다. 

ASGI 서버는 비동기 기반으로 설계된 프로토콜로 HTTP 뿐만 아니라 웹소켓, Server-sentEvents 등 실시간 통신과 비동기 작업을 위한 표준 인터페이스이다.  주로 저수준 네트워크 작업, 실시간 데이터 전송, 고성능 요구 등 특정한 제어가 필요한 경우 발생한다.

이러한 경우에는 scope receive send 를 직접 다루며 웹소켓 연결을 관리해야한다.=

 

FastAPI의 웹소켓은 기본적으로 텍스트 또는 바이너리 데이터 송수신을 지원하지만, 데이터를 패킷 구조로 해석하지 않기에, 세밀한 커스텀 데이터 포맷을 처리하기 위해서는 ASGI 레벨에서 직접 데이터 스트림을 다뤄야 한다

 

이러한 경우에서는 유비콘이나 하이퍼콘 같은 아스기서버를 바로 파이썬 asyncio 라이브러리와 사용해 작업하는게 더 나을 수도 있다.

 

import uvicorn
from starlette.websockets import WebSocket
async def app(scope, receive, send):
    websocket = WebSocket(scope, receive, send)
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        # Process data here...
        await websocket.send_text(f"Processed: {data}")
if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

 

 

2. FastAPI 는 단일 프로토콜에 대한 라우팅만 지원하기 때문에, 멀티플로토콜 작업시 asgi 작업 수준이 필요할 수도 있다. 

async def app(scope, receive, send):
    if scope["type"] == "http":
        await send({"type": "http.response.start", "status": 200})
        await send({"type": "http.response.body", "body": b"HTTP response"})
    elif scope["type"] == "websocket":
        websocket = WebSocket(scope, receive, send)
        await websocket.accept()
        await websocket.send_text("WebSocket response")

 

3. 대용량 스트리밍과 버퍼 조정이 필요한 경우가 발생할 수 있다.

FastAPI의 스트리밍 리스폰스는 편리하나, 세부적인 버퍼 크기 및 전송속도 조절이 어렵다.

 

with open("large_file.mp4", "rb") as file:
    while chunk := file.read(1024 * 1024):  # 1MB씩 전송
        await send({"type": "http.response.body", "body": chunk, "more_body": True})

 

4. SSE 및 실시간 데이터 푸시가 필요한 상황.

FastAPI 의 기본 라우팅은 요청 응답 기반으로 작동하지만, ASGI 레벨에서는 클라이언트 요청 없이도 이벤트를 푸시할 수 있는 구조를 제공한다. 패스트 에이피아이를 사용한다면 처음 연결될때만 의존성 주입등이 이루어지고 연결 후 주고 받는 메시지 등에서는 추가적 인증등을 매번 적용하기 어려울 것이다.

 

await send({"type": "http.response.start", "status": 200, "headers": [(b"content-type", b"text/event-stream")]})
while True:
    await asyncio.sleep(1)
    event = f"data: {datetime.now()}\n\n"
    await send({"type": "http.response.body", "body": event.encode("utf-8"), "more_body": True})

asgi 레벨에서는 지속적 데이터 스트림이 가능하다.

 

cors 를 어떻게 처리할 것인가. 

 

CORSmidddleware 를 통해 

allow_origins, allow_credentials, allow_methods, allow_headers 를 수정하여 처리할 수 있다.

 

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
origins = ["http://localhost:3000"]
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
@app.get("/")
async def main():
    return {"message": "Hello World"}