FastAPI + AWS S3

이번엔 FastAPI에서 만들어낸 오디오 파일을 S3에 저장하고 그 저장한 파일의 URL을 Spring Boot로 리턴하는 코드를 만들어야 한다.

전에 Spring boot와 S3에서 사용했던 S3 그대로 사용하자

프로젝트 구조

main.py

# -*- coding: utf-8 -*-
import os

from aiohttp import ClientError
from fastapi import FastAPI, File, UploadFile
from pathlib import Path

# ipynb 실행용
import nbformat
from nbconvert import PythonExporter
from nbconvert.preprocessors import ExecutePreprocessor

from typing import List

import boto3

import dotenv

dotenv.load_dotenv()

app = FastAPI()

# S3 연결 준비
client_s3 = boto3.client(
    's3',
    aws_access_key_id=os.getenv("CREDENTIALS_ACCESS_KEY"),
    aws_secret_access_key=os.getenv("CREDENTIALS_SECRET_KEY")
)

class ImageURLRequest(BaseModel):
    # meditationIdx: int
    images: List[str]

@app.post("/ai/text")
def ipynb(imageRequest: ImageURLRequest):
    audioUrl = []
    fileName = []

    # IPython 노트북 파일 경로 설정
    ipynb_file_path = "ipynb/image_chatgpt.ipynb"

    # IPython 노트북 파일을 읽기
    with open(ipynb_file_path, 'r', encoding='utf-8') as nb_file:
        notebook = nbformat.read(nb_file, as_version=4)

    print(imageRequest.images)

    # 매개변수 전달
    for image in imageRequest.images:
        for cell in notebook.cells:
            if cell.cell_type == 'code':
                cell.source = cell.source.replace("{{ imageUrl }}", image)

        # Python 코드로 변환
        python_exporter = PythonExporter()
        python_code, _ = python_exporter.from_notebook_node(notebook)

        # Python 코드 실행 및 결과 추출
        exec_preprocessor = ExecutePreprocessor(timeout=600)
        exec_preprocessor.preprocess(notebook, {'metadata': {'path': './'}})

        # 실행 결과 가져오기
        cell_outputs = notebook.cells[-1].outputs

        # 결과를 문자열로 가져옴
        result = ""
        for output in cell_outputs:
            if 'text' in output:
                result += output['text']
            elif 'data' in output and 'text/plain' in output['data']:
                result += output['data']['text/plain']

        # 명상용 텍스트 -> 음성 (미완성부분)

        file_path = "./audio/" + "파일명"

        # 파일 저장 (UploadFile -> 실제 저장할 파일). 로컬에 저장되어 있어야 S3에 저장할 수 있겠죠?
		# 그러기 위해선 로컬에 저장된 파일의 위치와 이름만 알면 됨!
         with file_path.open("wb") as f:
             f.write(UploadFile.file.read())

		# S3에 저장할 파일명을 담고있는 리스트
        fileName.append("파일명")

		# Spring boot로 리턴
    return {"audios": saveAudioAtS3(fileName)}

# Audio 파일을 S3에 저장하는 코드
def saveAudioAtS3(audio):
    audioUrl = []

    try:
        for a in audio:
			# 지정한 S3 버킷에 audio 파일 저장
            client_s3.upload_file(
                "./audio/"+a,
                os.getenv("S3_BUCKET"),
                "audio/" + a,
                ExtraArgs={'ContentType': 'audio/mp3'}
            )

            # 로컬에 저장되어있는 음성파일 삭제
            os.remove("./audio/" + a)
						
			# S3에 저장한 음성파일의 URL
            audioUrl.append(os.getenv("S3_URL") + "/" + a)

        return audioUrl
    except ClientError as e:
        print(f'Credential error => {e}')
    except Exception as e:
        print(f"Another error => {e}")

전체적인 흐름은 로컬에 저장하고 싶은 파일을 저장한 뒤, 저장한 파일의 경로와 파일명을 가지고 S3 버킷에 저장하는 흐름!

.env

CLARIFAI_API_KEY=
CHATGPT_API_KEY=

# S3 관련 정보
S3_BUCKET=
S3_URL=
CREDENTIALS_ACCESS_KEY=
CREDENTIALS_SECRET_KEY=

requirements.txt

fastapi
jupyter
uvicorn
openai
clarifai
nbformat
nbclient
nbconvert
python-dotenv
boto3
pydantic
ipython

프로젝트에 필요한 라이브러리!