FastAPI로 이미지 업로드

기존에 Flask로 POST 요청을 받는 API를 만들었다. 그런데 최근  여기저기 정보를 찾던 중 FastAPI가 종종 언급되는 것을 봤다. 도대체 왜 FastAPI에 열광하는지 궁금해서 기존 Flask 프로젝트를 FastAPI로 리팩토링 해보았다.

  • 속도: 비동기 처리를 지원해 이름처럼 빠른 실행 속도
  • 데이터 검증: 타입 힌트를 이용해 데이터를 쉽게 검증
  • 자동 문서: /docs에 자동으로 API 문서 생성
  • 쉬운 난이도: Python과 REST API에 익숙하다면 쉽게 사용할 수 있음

설치 및 실행

pip install fastapi[standard] 
pip install uvicorn[standard] 
pip install python-multipart

FastAPI를 사용하기 위해 fastapiuvicorn을 반드시 설치해야 한다. 데이터 검증을 위해서는 python-multipart를 설치해야 한다.

# main.py

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def main():
    return {"msg": "Hello"}
  • main.py에서 FastAPI()를 통해 app을 생성한다.
  • @데코레이터를 통해 URL을 연결한다.
  • async는 비동기 함수를 정의한다.
  • JSON(key-value) 형식으로 값을 반환한다.

기본적인 API 코드다. localhost:8000/로 접근했을 때 {"msg":"Hello"}를 반환한다.

uvicorn main:app --reload

서버를 실행하는 명령어로 --reload는 코드가 변경되었을 때 자동으로 reload 하는 옵션이다.

etc-image-0


파일 업로드

from fastapi import File, UploadFile

@app.post("/img/")
async def get_image(file: UploadFile = File()):
    name = file.filename
    type_ = file.content_type
    return {"name": name, "type": type_}
  • @post/upload/에 POST 요청을 보낸다.
  • file 파라미터로 파일 객체를 받아온다.
  • FastAPI의 타입 힌트는 타입을 강제한다. 즉, :UploadFile을 통해 file의 타입을 검증할 수 있다.

간단한 POST 요청을 보내는 API를 작성했다.

API 요청 보내기

FastAPI의 자동 문서를 제공한다는 장점이 있다. 127.0.0.1:8000/docs로 접근하면 문서를 볼 수 있다. 여기서 이미지를 업로드해 보자.

etc-image-1
blob

해당 API를 찾아 "Try it out"을 누르고, 이미지를 선택한 다음 "Execute"하면 POST 요청이 실행된다. 아래로 내려오면 response를 확인할 수 있다. 이미지가 잘 전달된 것을 볼 수 있다.

requests로 요청 보내기

POST 요청을 보내는 다양한 방법이 있지만 이번에는 Python 코드로 요청을 보내고 받아보자.

pip install requests
import requests

url = "http://127.0.0.1:8000/img/"
image_path = "sample.jpg"

with open(image_path, "rb") as image_file:
    # {Field-name: File-name, File-object, File-type}
    files = {"file": (image_path, image_file, "image/jpeg")}
    response = requests.post(url, files=files)

print("Status:", response.status_code)
print("Response:", response.json())

post로 POST 요청을 보내고 response를 받아온다. API에서 "file"이라는 파라미터를 받기 때문에 전달할 필드 이름도 "file"을 사용한다.

Status: 200
Response: {'name': 'sample.jpg', 'type': 'image/jpeg'}

docs에서 실험했듯이 결과를 잘 받아온다.

이미지 저장하기

import os

@app.post("/upload/")
async def save_image(file: UploadFile = File()):
    content = await file.read()
    image_dir = os.path.join(IMG_DIR, file.filename)
    with open(image_dir, "wb") as fp:
        fp.write(content)

POST한 이미지를 저장하는 예시다. 이미지는 /static 폴더 아래에 저장된다.


템플릿으로 보내기

root/
 |-- main.py
 |-- static
      |-- *.jpg
 |-- templates
      |-- *.html
pip install jinja2
from fastapi.templating import Jinja2Templates

templates = Jinja2Templates(directory="templates")

사용할 템플릿이 저장될 위치를 지정해 준다. 위 예시에서 템플릿(HTML) 파일은 /templates 아래에 저장된다. 

from fastapi import Request
from fastapi.responses import HTMLResponse

@app.post("/result/", response_class=HTMLResponse)
async def show_img(request: Request, file: UploadFile = File()):
    file_name = save_image(file) # 사용자 정의 함수
    img_url = "/".join((IMG_URL_PATH, file_name))
    return templates.TemplateResponse(
        "result.html",
        {
            "request": request,
            "img_src": img_url,
            "name": file_name,
        },
    )
  • response_class를 통해 JSON이 아닌 HTML을 반환한다고 알려준다. 
  • TemplateResponse에 HTML 파일과 전달할 값을 작성한다. 위 예시는 result.html을 보여준다.
  • img_url은 "/img/..."으로 파일 경로가 아닌 URL 경로를 사용한다. (아래 "이미지 경로" 참고.)
  • request, img_src, name은 템플릿에서 변수처럼 사용한다.

이미지 경로

FastAPI 앱이 이미지를 찾기 위해 경로를 mount 해주어야 한다.

from fastapi.staticfiles import StaticFiles

app.mount("/img", StaticFiles(directory="static"))
  • 파일은 127.0.0.1:8000/img로 접근한다. 문자열은 반드시 /로 시작해야 한다.
  • 파일의 실제 경로는 static이다. 파일은 static 아래에 저장된다.

템플릿 작성

<!DOCTYPE html>
<html lang="en">
  <head>
    <title>Image</title>
  </head>
  <body>
    <img src="{{ img_url }}" />
    <p>{{ name }}</p>
  </body>
</html>
  • {{ }}는 텍스트가 아닌 FastAPI에서 받아온 변수를 뜻한다.
  • 예시에서 img_url, name은 변수 이름 대신 변수 값으로 템플릿에 출력된다.
  • img_url은 "/img/..."으로 mount 한 경로부터 시작해야 한다.

(추가) URL, Query

이미지를 중심으로 설명했지만 URL이나 Query를 사용하는 방법도 있다.

@app.get("/id/{item_id}")
async def read_id(item_id: int = Path(ge=0, le=10)):
    return {"item_id": item_id}

# /id/3 -> 200 | {"item_id": 3}
# /id/12 -> 422 | ...
@app.get("/mul/")
async def get_query(n: int = 3, m: Union[int, None] = None):
    if m is None:
        return {"num": n}
    return {"num": n * m}


# /mul/?n=3&m=2 -> {"num":6}
# /mul/?n=5 -> {"num":5}
# /mul/ -> {"num":3}