FastAPI 베이스캠프

파일 업로드 및 처리

1. 파일 업로드 소개

파일 업로드는 많은 웹 애플리케이션에서 중요한 기능입니다. 사용자 프로필 사진 업로드, 문서 제출, 이미지 갤러리 등 다양한 상황에서 파일 업로드 기능이 필요합니다. FastAPI는 파이썬의 표준 라이브러리와 함께 파일 업로드를 쉽게 구현할 수 있는 기능을 제공합니다.

2. 단일 파일 업로드

FastAPI에서 단일 파일을 업로드하는 방법을 알아보겠습니다.

from fastapi import FastAPI, File, UploadFile
 
app = FastAPI()
 
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    return {"filename": file.filename}

from fastapi import FastAPI, File, UploadFile
 
app = FastAPI()
 
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    return {"filename": file.filename}

이 예제에서는 UploadFile과 File을 사용하여 파일 업로드를 처리합니다. UploadFile은 파일 이름, 내용 타입 등의 메타데이터에 접근할 수 있게 해줍니다.

3. 다중 파일 업로드

여러 파일을 동시에 업로드하는 방법도 있습니다.

from fastapi import FastAPI, File, UploadFile
from typing import List
 
app = FastAPI()
 
@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile] = File(...)):
    return {"filenames": [file.filename for file in files]}

from fastapi import FastAPI, File, UploadFile
from typing import List
 
app = FastAPI()
 
@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile] = File(...)):
    return {"filenames": [file.filename for file in files]}

이 예제에서는 여러 파일을 리스트로 받아 처리합니다.

4. 파일 저장하기

업로드된 파일을 서버에 저장하는 방법을 알아보겠습니다.

import shutil
from fastapi import FastAPI, File, UploadFile
 
app = FastAPI()
 
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    with open(f"uploaded_{file.filename}", "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    return {"filename": file.filename}

import shutil
from fastapi import FastAPI, File, UploadFile
 
app = FastAPI()
 
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    with open(f"uploaded_{file.filename}", "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    return {"filename": file.filename}

이 예제에서는 shutil.copyfileobj를 사용하여 업로드된 파일을 서버의 로컬 디렉토리에 저장합니다.

5. 파일 유효성 검사

업로드된 파일의 크기나 타입을 검증하는 것은 중요한 보안 조치입니다.

from fastapi import FastAPI, File, UploadFile, HTTPException
 
app = FastAPI()
 
ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg", "gif"}
MAX_FILE_SIZE = 1_000_000  # 1 MB
 
def allowed_file(filename: str):
    return "." in filename and filename.rsplit(".", 1)[1].lower() in ALLOWED_EXTENSIONS
 
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    if not allowed_file(file.filename):
        raise HTTPException(status_code=400, detail="File type not allowed")
 
    content = await file.read()
    if len(content) > MAX_FILE_SIZE:
        raise HTTPException(status_code=400, detail="File too large")
 
    # 파일 처리 로직...
 
    return {"filename": file.filename}

from fastapi import FastAPI, File, UploadFile, HTTPException
 
app = FastAPI()
 
ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg", "gif"}
MAX_FILE_SIZE = 1_000_000  # 1 MB
 
def allowed_file(filename: str):
    return "." in filename and filename.rsplit(".", 1)[1].lower() in ALLOWED_EXTENSIONS
 
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    if not allowed_file(file.filename):
        raise HTTPException(status_code=400, detail="File type not allowed")
 
    content = await file.read()
    if len(content) > MAX_FILE_SIZE:
        raise HTTPException(status_code=400, detail="File too large")
 
    # 파일 처리 로직...
 
    return {"filename": file.filename}

이 예제에서는 파일 확장자와 크기를 검사하여 허용된 파일만 처리합니다.

6. 파일 스트리밍

대용량 파일을 효율적으로 처리하기 위해 스트리밍을 사용할 수 있습니다.

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import StreamingResponse
import io
 
app = FastAPI()
 
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    return StreamingResponse(io.BytesIO(contents), media_type=file.content_type)

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import StreamingResponse
import io
 
app = FastAPI()
 
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    return StreamingResponse(io.BytesIO(contents), media_type=file.content_type)

이 예제에서는 업로드된 파일을 읽어서 바로 스트리밍 응답으로 반환합니다.

연습문제

이미지 파일만 허용하는 업로드 엔드포인트를 만들어보세요. 업로드된 이미지의 크기(폭과 높이)를 반환해야 합니다. (힌트: Pillow 라이브러리를 사용할 수 있습니다)
CSV 파일을 업로드 받아 그 내용을 JSON 형식으로 반환하는 엔드포인트를 구현해보세요.
업로드된 파일의 MD5 해시를 계산하여 반환하는 엔드포인트를 만들어보세요.
여러 파일을 한 번에 업로드 받아 ZIP 파일로 압축하여 반환하는 엔드포인트를 구현해보세요.