Fastapi: 프로젝트로 바로 적용하는 학습(1)

개요: 특정 스타트업의 아이템 추천시스템 구축

로직: streamlit 고객 입력(프론트) -> fastapi(백엔드, db쿼리 전달) -> db ->fastapi 모델서빙 -> 스트림릿

---

FastApi <-> mysql 연결 (local) #sqlalchemy을 통해 파이썬으로 연동 가능

1) 엔진 생성 - 세션생성 - Base 모델 선언(입력 데이터를 json형식, mysql 타입으로 가공하기 위한) - 엔드포인트 설정

 나의 경우, 데이터가 이미 쌓여있을 것이라 가정하고, mysql에 user_input 테이블을 미리 만들어두었다.

from fastapi import FastAPI
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, String, Integer, TEXT
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from enum import Enum
from typing import List, Optional
#import mysql.connector
#from mysql.connector import Error

# FastAPI 인스턴스 생성
app = FastAPI()

# MySQL 데이터베이스 연결 설정
DATABASE_URL = "mysql+mysqlconnector://사용자이름:비밀번호@호스트/db이름"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# SQLAlchemy 모델 정의
Base = declarative_base()

class UserInput(Base):
    __tablename__ = "user_input"
    id = Column(Integer, primary_key=True, index=True)
    customer_id = Column(String, index=True)
    password = Column(String)
    name = Column(String)
    gender = Column(String)
    age = Column(Integer)
    child_age = Column(Integer)
    child_gender = Column(String)
    parenting_concerns = Column(TEXT)

# Pydantic 모델 정의 (FastAPI의 request body를 검증하기 위해)
class GenderEnum(str, Enum):
    Male = "Male"
    Female = "Female"

class UserInputRequest(BaseModel):
    customer_id: str
    password: Optional[str]
    name: str
    gender: GenderEnum
    age: int
    child_age: int
    child_gender: GenderEnum
    parenting_concerns: List[str]

# FastAPI 엔드포인트 정의
@app.post("/submit_user_input")
def submit_user_input(user_input: UserInputRequest):
    # 입력 받은 데이터를 SQLAlchemy 모델에 매핑
    db_user_input = UserInput(
        customer_id=user_input.customer_id,
        password=user_input.password,
        name=user_input.name,
        gender=user_input.gender.value,
        age=user_input.age,
        child_age=user_input.child_age,
        child_gender=user_input.child_gender.value,
        parenting_concerns=",".join(user_input.parenting_concerns)
    )

    # 데이터베이스에 세션을 만들어서 데이터 추가
    db = SessionLocal()

    # 이미 존재하는 테이블에 데이터 추가
    db.execute(
        UserInput.__table__.insert().values(
            customer_id=db_user_input.customer_id,
            password=db_user_input.password,
            name=db_user_input.name,
            gender=db_user_input.gender,
            age=db_user_input.age,
            child_age=db_user_input.child_age,
            child_gender=db_user_input.child_gender,
            parenting_concerns=db_user_input.parenting_concerns
        )
    )

    # 커밋
    db.commit()

    return {"message": "User input submitted successfully"}