본문 바로가기

웹 프레임워크/FastAPI

FastAPI - 12 (파일 업로드 / 파일 다운로드)

업로드 된 파일 저장

예시 코드

from fastapi import FastAPI, UploadFile, File, Form
from fastapi.responses import HTMLResponse
import shutil
import os

app = FastAPI()

os.makedirs("uploads", exist_ok=True)

@app.get("/", response_class=HTMLResponse)
async def get_upload_form():
    return """
    <html>
        <head>
            <title>Upload File</title>
        </head>
        <body>
            <h1>Upload File with Extra Fields</h1>
            <form action="/uploadfile/" method="post" enctype="multipart/form-data">
                <input type="file" name="file"><br>
                <input type="text" name="item_id" placeholder="Item ID"><br>
                <input type="text" name="description" placeholder="Description"><br>
                <input type="submit" value="Upload">
            </form>
        </body>
    </html>
    """

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...), 
                             item_id: str = Form(...), 
                             description: str = Form(...)):
    file_location = f"uploads/{file.filename}"
    with open(file_location, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)

    return {
        "info": f"file '{file.filename}' saved at '{file_location}'",
        "item_id": item_id,
        "description": description
    }

주석이(설명) 포함 된 코드

# 파일 요청 처리를 위한 File, UploadFile import
# Form data 처리를 위한 Form import
from fastapi import FastAPI, UploadFile, File, Form
# HTML 화면을 위한 HTMLResponse import
from fastapi.responses import HTMLResponse
# 파일 저장을 위한 shutil import
import shutil
# 시스템 디렉토리 생성을 위한 os import
import os

app = FastAPI()

# uploads 디렉토리가 없으면 생성
# exist_ok=True를 사용하여 디렉토리가 이미 존재하는 경우 오류를 발생시키지 않습니다.
os.makedirs("uploads", exist_ok=True)

# request를 위한 HTML 화면
# response_class=HTMLResponse를 사용하여 HTML 코드를 반환합니다.
# enctype="multipart/form-data"를 사용하여 파일 업로드를 지원합니다.
@app.get("/", response_class=HTMLResponse)
async def get_upload_form():
    return """
    <html>
        <head>
            <title>Upload File</title>
        </head>
        <body>
            <h1>Upload File with Extra Fields</h1>
            <form action="/uploadfile/" method="post" enctype="multipart/form-data">
                <input type="file" name="file"><br>
                <input type="text" name="item_id" placeholder="Item ID"><br>
                <input type="text" name="description" placeholder="Description"><br>
                <input type="submit" value="Upload">
            </form>
        </body>
    </html>
    """

# 클라이언트로 부터 요청된 파일을 저장합니다.
@app.post("/uploadfile/")
# 함수 매개변수 file을 선언했으며 유형은 UploadFile로 업로드된 파일에 대한 정보와 메서드를 포함하고 있습니다.
# UploadFile 은 파일의 내용을 비동기적으로 읽고 쓸 수 있게 해주며, 파일명, MIME 타입과 같은 메타데이터도 제공합니다.
# File(...)은 업로드된 파일을 UploadFile 객체로 받게 해주는 FastAPI의 함수입니다.
# File(...)의 ... 은 매개변수가 필수임을 나타냅니다. 반드시 파일이 업로드 되어야 합니다. 그렇지 않으면 오류가 발생합니다.
# item_id와 description은 Form(...)을 사용하여 Form으로 요청한 text를 받습니다.
async def create_upload_file(file: UploadFile = File(...), 
                             item_id: str = Form(...), 
                             description: str = Form(...)):
    # file 개체로 업로드 된 파일의 속성을 확인할 수 있습니다.
    # file_location 변수에 업로드 된 파일의 경로(파일명 포함)를 지정합니다.
    file_location = f"uploads/{file.filename}"
    # open() 함수를 사용하여 파일을 엽니다.
    with open(file_location, "wb") as buffer:
        # file.file은 업로되 된 파일 binary를 읽습니다.
        # shutil.copyfileobj() 함수를 사용하여 file.file에서 읽은 binary를 buffer에 씁니다.
        shutil.copyfileobj(file.file, buffer)

    return {
        "info": f"file '{file.filename}' saved at '{file_location}'",
        "item_id": item_id,
        "description": description
    }

테스트

http://localhost:8000/

위의 URL로 접속하면 아래와 같이 파일과 항목 2개를 요청 할 수 화면이 나옵니다.

업로드 할 파일을 선택하고 폼 데이터(item, description)을 입력합니다.

Upload 버튼을 클릭하면 업로드 된 파일의 경로와 폼 데이터(item, description)을 확인 할 수 있습니다.

uploads 디렉토리에 업로드 된 파일이 저장되어 있는 것을 확인 할 수 있습니다.

참고로 저는 원격 서버를 사용 하고 있어 FTP로 접속 한 화면을 캡쳐 했습니다.

파일 다운로드

파일 다운로드 예시 코드

...
### 기존 코드 ###

# 파일 다운로드를 위한 FileResponse import
from fastapi.responses import FileResponse

@app.get("/download/{filename}")
async def download_file(filename: str):
    # 현재 디렉토리의 경로를 가져옵니다.
    current_directory = os.getcwd()
    # 현재 디렉토리의 경로ㅗ아 uploads 디렉토리의 경로와 파일명을 합쳐 파일의 전체 경로를 만듭니다.
    file_path = os.path.join(current_directory, "uploads", filename)
    # FileResponse를 사용하여 파일을 다운로드 합니다.
    return FileResponse(path=file_path, filename=filename)

다운로드 테스트

업로드에서 테스트 했던 파일(fast-12_01.png)을 다운로드 합니다.

http://localhost:8000/download/fast-12_01.png

위의 URL로 접속하면 fast-12_01.png 파일이 다운로드 되는 것을 확인 할 수 있습니다.