WeniVooks

검색

FastAPI 베이스캠프

DB 구성 및 static 파일 서빙

1. DB 구성

앞서 배웠던 sqlite3를 사용하여 데이터베이스를 구성해보겠습니다. sqlalchemy와 pydantic 모듈은 05-2 챕터에서 설치하였습니다. 혹시 설치가 되어있지 않다면 아래 명령어를 사용하여 설치해주세요. 뒤에서 사용할 코드까지 모두 포함되어 있습니다.

pip install sqlalchemy pydantic
pip install sqlalchemy pydantic
from fastapi import FastAPI, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
import datetime
 
# Database 설정
SQLALCHEMY_DATABASE_URL = "sqlite:///./blogs.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
 
 
# Blog 모델 정의
class BlogModel(Base):
    __tablename__ = "blogs"
 
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, nullable=False)
    content = Column(String, nullable=False)
    author = Column(String, nullable=False)
    created_at = Column(String, nullable=False)
    updated_at = Column(String, nullable=False)
 
 
# 데이터베이스 테이블 생성
Base.metadata.create_all(bind=engine)
 
app = FastAPI()
 
# CORS 설정
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
 
app.mount("/static", StaticFiles(directory="static"), name="static")
 
 
# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
 
 
# Pydantic 모델
class BlogCreate(BaseModel):
    title: str
    content: str
 
 
class Blog(BaseModel):
    id: int
    title: str
    content: str
    author: str
    created_at: str
    updated_at: str
 
    class Config:
        from_attributes = True
 
 
# 초기 데이터 삽입 함수
def init_db():
    db = SessionLocal()
    try:
        # 데이터가 없을 경우에만 초기 데이터 삽입
        if not db.query(BlogModel).first():
            initial_data = [
                BlogModel(
                    title="Hello",
                    content="World",
                    author="admin",
                    created_at="2025-01-06",
                    updated_at="2025-01-06",
                ),
                BlogModel(
                    title="FastAPI",
                    content="Python",
                    author="admin",
                    created_at="2025-01-07",
                    updated_at="2025-01-07",
                ),
                BlogModel(
                    title="Django",
                    content="Python",
                    author="admin",
                    created_at="2025-01-08",
                    updated_at="2025-01-08",
                ),
            ]
            for data in initial_data:
                db.add(data)
            db.commit()
    except Exception as e:
        print(f"Error initializing database: {e}")
        db.rollback()
    finally:
        db.close()
 
 
# 애플리케이션 시작 시 초기 데이터 삽입
init_db()
 
 
# 블로그 글 목록 조회
@app.get("/blogs", response_model=list[Blog])
def read_blogs(db: Session = Depends(get_db)):
    blogs = db.query(BlogModel).order_by(BlogModel.id.desc()).all()
    return blogs
 
 
# 블로그 글 생성
@app.post("/blogs", response_model=Blog)
def create_blog(blog_create_data: BlogCreate, db: Session = Depends(get_db)):
    now = datetime.datetime.now()
    created_at = now.strftime("%Y-%m-%d")
    print(blog_create_data)
 
    new_blog = BlogModel(
        title=blog_create_data.title,
        content=blog_create_data.content,
        author="admin",
        created_at=created_at,
        updated_at=created_at,
    )
 
    db.add(new_blog)
    db.commit()
    db.refresh(new_blog)
 
    return new_blog
from fastapi import FastAPI, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
import datetime
 
# Database 설정
SQLALCHEMY_DATABASE_URL = "sqlite:///./blogs.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
 
 
# Blog 모델 정의
class BlogModel(Base):
    __tablename__ = "blogs"
 
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, nullable=False)
    content = Column(String, nullable=False)
    author = Column(String, nullable=False)
    created_at = Column(String, nullable=False)
    updated_at = Column(String, nullable=False)
 
 
# 데이터베이스 테이블 생성
Base.metadata.create_all(bind=engine)
 
app = FastAPI()
 
# CORS 설정
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
 
app.mount("/static", StaticFiles(directory="static"), name="static")
 
 
# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
 
 
# Pydantic 모델
class BlogCreate(BaseModel):
    title: str
    content: str
 
 
class Blog(BaseModel):
    id: int
    title: str
    content: str
    author: str
    created_at: str
    updated_at: str
 
    class Config:
        from_attributes = True
 
 
# 초기 데이터 삽입 함수
def init_db():
    db = SessionLocal()
    try:
        # 데이터가 없을 경우에만 초기 데이터 삽입
        if not db.query(BlogModel).first():
            initial_data = [
                BlogModel(
                    title="Hello",
                    content="World",
                    author="admin",
                    created_at="2025-01-06",
                    updated_at="2025-01-06",
                ),
                BlogModel(
                    title="FastAPI",
                    content="Python",
                    author="admin",
                    created_at="2025-01-07",
                    updated_at="2025-01-07",
                ),
                BlogModel(
                    title="Django",
                    content="Python",
                    author="admin",
                    created_at="2025-01-08",
                    updated_at="2025-01-08",
                ),
            ]
            for data in initial_data:
                db.add(data)
            db.commit()
    except Exception as e:
        print(f"Error initializing database: {e}")
        db.rollback()
    finally:
        db.close()
 
 
# 애플리케이션 시작 시 초기 데이터 삽입
init_db()
 
 
# 블로그 글 목록 조회
@app.get("/blogs", response_model=list[Blog])
def read_blogs(db: Session = Depends(get_db)):
    blogs = db.query(BlogModel).order_by(BlogModel.id.desc()).all()
    return blogs
 
 
# 블로그 글 생성
@app.post("/blogs", response_model=Blog)
def create_blog(blog_create_data: BlogCreate, db: Session = Depends(get_db)):
    now = datetime.datetime.now()
    created_at = now.strftime("%Y-%m-%d")
    print(blog_create_data)
 
    new_blog = BlogModel(
        title=blog_create_data.title,
        content=blog_create_data.content,
        author="admin",
        created_at=created_at,
        updated_at=created_at,
    )
 
    db.add(new_blog)
    db.commit()
    db.refresh(new_blog)
 
    return new_blog

2. static 파일 서빙

05-4 챕터에서는 MPA로 블로그를 구현하였습니다. 이렇게 하면 live server와 FastAPI 서버를 동시에 실행해야 하는 번거로움이 있습니다. 이를 해결하기 위해 FastAPI에서 제공하는 StaticFiles 클래스를 사용하여 static 파일을 서빙할 수 있습니다.

app.mount("/static", StaticFiles(directory="static"), name="static")
app.mount("/static", StaticFiles(directory="static"), name="static")

위 코드를 추가하면 static 디렉토리에 있는 파일을 /static 경로로 접근할 수 있습니다. 이제 static 디렉토리를 생성하고 모든 .html 파일을 이동시킵니다. 여기서 이동한 파일들의 경로는 http://127.0.0.1:8000/static/파일명.html로 접근할 수 있습니다. 따라서 live server 경로로 되어 있던 http://127.0.0.1:5500을 위와 같이 수정하셔야 합니다.

또한 위와 같이 설정했을 경우 최상위 폴더 아래 static이라는 폴더가 없을 경우 애러가 납니다. 따라서 폴더도 미리 구성해야 합니다. 만약 이러한 설정이 번거롭거나 견고하지 못하다고 생각되면 os 모듈을 사용하여 다음과 같이 처음에 폴더가 있는지를 검사하여 없을 경우 생성하도록 할 수 있습니다.

import os
 
if not os.path.exists("static"):
    os.makedirs("static")
import os
 
if not os.path.exists("static"):
    os.makedirs("static")

3. 실행

아래 코드로 FastAPI 서버를 실행하여 블로그를 확인해보세요. 이제 라이브서버로 실행할 필요는 없습니다.

uvicorn main:app --reload
uvicorn main:app --reload
5.4 MPA로 블로그 구현하기5.6 인증 구현하기