Almost there

FastAPI PostgreSQL

March 28, 2023

이번 글은 FastAPI 프레임워크에서 PostgreSQL을 연동하여 API를 처음 만드는 개발자를 대상으로 빠르게 구현의 목적이 있습니다. 상세 설명이 필요하시다면 다른 곳을 참조 하시는걸 추천 드립니다.

설명이 부족하여 이해의 어려움이 있다면, Github 저장소에서 소스 전체를 확인 해보시면 도움이 될 것 입니다. (이번 글의 개발 환경은 macOS에서 진행 되었습니다.)

사전 준비 사항

CREATE SEQUENCE board_board_no_seq;

CREATE TABLE IF NOT EXISTS public.board
(
    board_no bigint NOT NULL DEFAULT nextval('board_board_no_seq'::regclass),
    title character varying(200) COLLATE pg_catalog."default" NOT NULL,
    contents text COLLATE pg_catalog."default" NOT NULL,
    writer character varying(50) COLLATE pg_catalog."default" NOT NULL,
    view_count integer NOT NULL,
    link_url character varying(200) COLLATE pg_catalog."default" NOT NULL,
    create_date timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
    update_date timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
    CONSTRAINT board_pkey PRIMARY KEY (board_no)
)

CREATE OR REPLACE FUNCTION update_timestamp()
RETURNS TRIGGER AS $$
BEGIN
  NEW.update_date = CURRENT_TIMESTAMP;
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER board_update_timestamp_trigger
BEFORE UPDATE ON public.board
FOR EACH ROW
EXECUTE PROCEDURE update_timestamp();

FastAPI 설치

터미널에서 프로젝트를 생성합니다.

$ mkdir fastapi-pg

위에서 생성한 프로젝트 경로로 이동합니다.

$ cd fastapi-pg

터미널에서 “pip” 명령어로 fastapi를 설치 합니다.

$ pip install fastapi

fastapi pg 01

uvicorn 웹서버를 설치 합니다.

$ pip install "uvicorn[standard]"

fastapi pg 02

Hello World

VS Code에서 “/main.py” 파일을 생성 후 아래 코드를 추가 합니다.

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def hello():
    return {"Hello": "World"}

웹 서버를 실행합니다.

$ uvicorn main:app --reload

브라우저에서 http://127.0.0.1:8000/docs 에 접속합니다.

fastapi pg 03

“Hello” 클릭해서 펼친 후 “Try it out” 버튼을 클릭합니다.

fastapi pg 04

“Execute” 버튼을 클릭합니다.

fastapi pg 05

API를 실행 후 “Response body” 결과를 확인 합니다.

fastapi pg 06

Database 설정

PostgreSQL 라이브러리 데이터베이스 연동 관련 라이브러리를 설치 합니다.

$ pip install SQLAlchemy
$ pip install python-dotenv
$ pip install pydantic
$ pip install psycopg2

“/config/.env” 파일을 생성 후 아래의 형식으로 데이터베이스 접속 정보를 입력합니다.

host=localhost
port=5432
user=아이디
password=비밀번호
db=데이터베이스명
dbtype=postgresql

“/db/database.py” 파일을 생성 후 아래 코드를 추가합니다.

import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from dotenv import load_dotenv

# load .env
load_dotenv(dotenv_path="config/.env", verbose=True)

host = os.environ.get('host')
port = os.environ.get('port')
user = os.environ.get('user')
password = os.environ.get('password')
db = os.environ.get('db')
dbtype = os.environ.get('dbtype')

SQLALCHEMY_DATABASE_URL = f"{dbtype}://{user}:{password}@{host}:{port}/{db}"

engine = create_engine(SQLALCHEMY_DATABASE_URL)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

“/db/connection.py” 파일을 생성 후 아래 코드를 추가합니다.

from db.database import SessionLocal

def get_db():
	db = SessionLocal()

	try:
		yield db
	finally:
		db.close()

“/models.py” 파일을 생성 후 아래 코드를 추가합니다.

from sqlalchemy import Column, String, Integer, TEXT, DATETIME
from db.database import Base
from sqlalchemy.sql import func

class Board(Base):
    __tablename__ = "board"

    board_no = Column(Integer, primary_key=True, index=True)
    title = Column(String(200))
    contents = Column(TEXT)
    writer = Column(String(50))
    view_count = Column(Integer)
    link_url = Column(String(200))
    create_date = Column(DATETIME(timezone=True), server_default=func.now())
    update_date = Column(DATETIME(timezone=True), server_default=func.now())

“/schemas.py” 파일을 생성 후 아래 코드를 추가합니다.

from pydantic import BaseModel

class Item(BaseModel):
    board_no: int | None = None
    title: str
    contents: str
    writer: str | None = None
    view_count: int | None = None
    link_url: str | None = None

    class Config:
        orm_mode = True

“/main.py” 파일에 “sqlalchemy, db.connection, pydantic, models, schemas” 를 import 하고, Response 함수를 추가 합니다.

from fastapi import FastAPI
from fastapi.params import Depends
from sqlalchemy.orm import Session
from db.connection import get_db
from pydantic import ValidationError
from models import Board
import schemas

def Response(status, message, data):
    return {
        "status": status,
        "message": message,
        "data": data
    }

app = FastAPI()

@app.get("/")
def hello():
    return {"Hello": "World"}

등록 API

“/main.py” 파일에 “create_board” 함수를 추가합니다.

from fastapi import FastAPI
from fastapi.params import Depends
from sqlalchemy.orm import Session
from db.connection import get_db
from pydantic import ValidationError
from models import Board
import schemas

def Response(status, message, data):
    return {
        "status": status,
        "message": message,
        "data": data
    }

app = FastAPI()

@app.get("/")
def hello():
    return {"Hello": "World"}


@app.post("/board")
def create_board(item: schemas.Item, db: Session = Depends(get_db)):
    try:
        board = Board()
        board.title =  item.title
        board.contents =  item.contents
        board.writer =  item.writer
        board.view_count =  item.view_count
        board.link_url =  item.link_url

        db.add(board)
        db.flush()

        db.refresh(board, attribute_names=['board_no'])
        data = {"board_no": board.board_no}
        db.commit()

        status = True
        message = "Board added successfully."
    except ValidationError as e:
        status = False
        data = None
        message = e

    return Response(status, message, data)

브라우저에서 http://127.0.0.1:8000/docs 에 접속합니다.

fastapi pg 07

“Create Board” 를 클릭해서 펼친 후 “Try it out” 버튼을 클릭합니다.

fastapi pg 08

“Request Body” 에 아래의 JSON 형식으로 값을 입력 후 “Execute” 버튼을 클릭합니다.

{
  "title": "제목",
  "contents": "내용",
  "writer": "작성자",
  "view_count": 1,
  "link_url": "https://"
}

fastapi pg 09

API를 실행 후 “Response body” 에서 결과를 확인 합니다.

fastapi pg 10

목록조회 API

“/main.py” 파일에 “getboardall” 함수를 추가합니다.

from fastapi import FastAPI
from fastapi.params import Depends
from sqlalchemy.orm import Session
from db.connection import get_db
from pydantic import ValidationError
from models import Board
import schemas

def Response(status, message, data):
    return {
        "status": status,
        "message": message,
        "data": data
    }

app = FastAPI()

@app.get("/")
def hello():
    return {"Hello": "World"}


@app.get("/board")
def get_board_all(db: Session = Depends(get_db)):
    datas = db.query(Board).all()

    status = True
    message = "Board retrieved successfully"

    if len(datas) <= 0:
        status = False
        message = "Board not found"

    return Response(status, message, datas)


@app.post("/board")
def create_board(item: schemas.Item, db: Session = Depends(get_db)):
    try:
        board = Board()
        board.title =  item.title
        board.contents =  item.contents
        board.writer =  item.writer
        board.view_count =  item.view_count
        board.link_url =  item.link_url

        db.add(board)
        db.flush()

        db.refresh(board, attribute_names=['board_no'])
        data = {"board_no": board.board_no}
        db.commit()

        status = True
        message = "Board added successfully."
    except ValidationError as e:
        status = False
        data = None
        message = e

    return Response(status, message, data)

브라우저에서 http://127.0.0.1:8000/docs 에 접속합니다.

fastapi pg 11

“Get Board All” 를 클릭해서 펼친 후 “Try it out” 버튼을 클릭합니다.

fastapi pg 12

“Execute” 버튼을 클릭합니다.

fastapi pg 13

API를 실행 후 “Response body” 에서 결과를 확인 합니다.

fastapi pg 14

조회 API

“/main.py” 파일에 “getboardbyboardno” 함수를 추가합니다.

from fastapi import FastAPI
from fastapi.params import Depends
from sqlalchemy.orm import Session
from db.connection import get_db
from pydantic import ValidationError
from models import Board
import schemas

def Response(status, message, data):
    return {
        "status": status,
        "message": message,
        "data": data
    }

app = FastAPI()

@app.get("/")
def hello():
    return {"Hello": "World"}


@app.get("/board")
def get_board_all(db: Session = Depends(get_db)):
    datas = db.query(Board).all()

    status = True
    message = "Board retrieved successfully"

    if len(datas) <= 0:
        status = False
        message = "Board not found"

    return Response(status, message, datas)


@app.get("/board/{board_no}")
def get_board_by_board_no(board_no: str, db: Session = Depends(get_db)):
    datas = db.query(Board).filter(Board.board_no == board_no).all()

    status = True
    message = "Board retrieved successfully"

    if len(datas) <= 0:
        status = False
        message = "Board not found"

    return Response(status, message, datas)


@app.post("/board")
def create_board(item: schemas.Item, db: Session = Depends(get_db)):
    try:
        board = Board()
        board.title =  item.title
        board.contents =  item.contents
        board.writer =  item.writer
        board.view_count =  item.view_count
        board.link_url =  item.link_url

        db.add(board)
        db.flush()

        db.refresh(board, attribute_names=['board_no'])
        data = {"board_no": board.board_no}
        db.commit()

        status = True
        message = "Board added successfully."
    except ValidationError as e:
        status = False
        data = None
        message = e

    return Response(status, message, data)

브라우저에서 http://127.0.0.1:8000/docs 에 접속합니다.

fastapi pg 15

“Get Board By Board No” 를 클릭해서 펼친 후 “Try it out” 버튼을 클릭합니다.

fastapi pg 16

“Parameters” 의 “boardno” 에 조회할 boardno를 입력 후 “Execute” 버튼을 클릭합니다.

fastapi pg 17

API를 실행 후 “Response body” 에서 결과를 확인 합니다.

fastapi pg 18

수정 API

“/main.py” 파일에 “update_board” 함수를 추가합니다.

from fastapi import FastAPI
from fastapi.params import Depends
from sqlalchemy.orm import Session
from db.connection import get_db
from pydantic import ValidationError
from models import Board
import schemas

def Response(status, message, data):
    return {
        "status": status,
        "message": message,
        "data": data
    }

app = FastAPI()

@app.get("/")
def hello():
    return {"Hello": "World"}


@app.get("/board")
def get_board_all(db: Session = Depends(get_db)):
    datas = db.query(Board).all()

    status = True
    message = "Board retrieved successfully"

    if len(datas) <= 0:
        status = False
        message = "Board not found"

    return Response(status, message, datas)


@app.get("/board/{board_no}")
def get_board_by_board_no(board_no: str, db: Session = Depends(get_db)):
    datas = db.query(Board).filter(Board.board_no == board_no).all()

    status = True
    message = "Board retrieved successfully"

    if len(datas) <= 0:
        status = False
        message = "Board not found"

    return Response(status, message, datas)


@app.post("/board")
def create_board(item: schemas.Item, db: Session = Depends(get_db)):
    try:
        board = Board()
        board.title =  item.title
        board.contents =  item.contents
        board.writer =  item.writer
        board.view_count =  item.view_count
        board.link_url =  item.link_url

        db.add(board)
        db.flush()

        db.refresh(board, attribute_names=['board_no'])
        data = {"board_no": board.board_no}
        db.commit()

        status = True
        message = "Board added successfully."
    except ValidationError as e:
        status = False
        data = None
        message = e

    return Response(status, message, data)


@app.put("/board")
def update_board(item: schemas.Item, db: Session = Depends(get_db)):
    try:
        is_updated = db.query(Board).filter(Board.board_no == item.board_no).update({
            Board.title: item.title,
            Board.contents: item.contents,
            Board.writer: item.writer,
            Board.view_count: item.view_count,
            Board.link_url: item.link_url
        }, synchronize_session=False)

        db.flush()
        db.commit()

        status = True
        message = "Board updated successfully"

        if is_updated == 1:
            data = db.query(Board).filter(Board.board_no == item.board_no).one()
        elif is_updated == 0:
            message = "Board not updated. No product found with this board_no :" + \
                str(item.board_no)
            status = False
            data = None
    except Exception as e:
        status = False
        data = None
        message = e

    return Response(status, message, data)

브라우저에서 http://127.0.0.1:8000/docs 에 접속합니다.

fastapi pg 19

“Update Board” 를 클릭해서 펼친 후 “Try it out” 버튼을 클릭합니다.

fastapi pg 20

“Request Body” 에 아래의 JSON 형식으로 값을 입력 후 “Execute” 버튼을 클릭합니다.

{
  "board_no": 1,
  "title": "제목 수정",
  "contents": "내용 수정",
  "writer": "작성자 수정",
  "view_count": 2,
  "link_url": "https://"
}

fastapi pg 21

API를 실행 후 “Response body” 에서 결과를 확인 합니다.

fastapi pg 22

삭제 API

“/main.py” 파일에 “delete_board” 함수를 추가합니다.

from fastapi import FastAPI
from fastapi.params import Depends
from sqlalchemy.orm import Session
from db.connection import get_db
from pydantic import ValidationError
from models import Board
import schemas

def Response(status, message, data):
    return {
        "status": status,
        "message": message,
        "data": data
    }

app = FastAPI()

@app.get("/")
def hello():
    return {"Hello": "World"}


@app.get("/board")
def get_board_all(db: Session = Depends(get_db)):
    datas = db.query(Board).all()

    status = True
    message = "Board retrieved successfully"

    if len(datas) <= 0:
        status = False
        message = "Board not found"

    return Response(status, message, datas)


@app.get("/board/{board_no}")
def get_board_by_board_no(board_no: str, db: Session = Depends(get_db)):
    datas = db.query(Board).filter(Board.board_no == board_no).all()

    status = True
    message = "Board retrieved successfully"

    if len(datas) <= 0:
        status = False
        message = "Board not found"

    return Response(status, message, datas)


@app.post("/board")
def create_board(item: schemas.Item, db: Session = Depends(get_db)):
    try:
        board = Board()
        board.title =  item.title
        board.contents =  item.contents
        board.writer =  item.writer
        board.view_count =  item.view_count
        board.link_url =  item.link_url

        db.add(board)
        db.flush()

        db.refresh(board, attribute_names=['board_no'])
        data = {"board_no": board.board_no}
        db.commit()

        status = True
        message = "Board added successfully."
    except ValidationError as e:
        status = False
        data = None
        message = e

    return Response(status, message, data)


@app.put("/board")
def update_board(item: schemas.Item, db: Session = Depends(get_db)):
    try:
        is_updated = db.query(Board).filter(Board.board_no == item.board_no).update({
            Board.title: item.title,
            Board.contents: item.contents,
            Board.writer: item.writer,
            Board.view_count: item.view_count,
            Board.link_url: item.link_url
        }, synchronize_session=False)

        db.flush()
        db.commit()

        status = True
        message = "Board updated successfully"

        if is_updated == 1:
            data = db.query(Board).filter(Board.board_no == item.board_no).one()
        elif is_updated == 0:
            message = "Board not updated. No product found with this board_no :" + \
                str(item.board_no)
            status = False
            data = None
    except Exception as e:
        status = False
        data = None
        message = e

    return Response(status, message, data)


@app.delete("/board/{board_no}")
def delete_board(board_no: int, db: Session = Depends(get_db)):
    data = None

    try:
        status = True

        is_deleted = db.query(Board).filter(Board.board_no == board_no).delete()
        db.commit()

        if is_deleted == 1:
            message = "Board deleted successfully"
        else:
            message = "Board not updated. No product found with this board_no :" + \
                str(board_no)
    except Exception as e:
        status = False
        message = e

    return Response(status, message, data)

브라우저에서 http://127.0.0.1:8000/docs 에 접속합니다.

fastapi pg 23

“Delete Board” 를 클릭해서 펼친 후 “Try it out” 버튼을 클릭합니다.

fastapi pg 24

“Parameters” 의 “boardno” 에 삭제할 boardno 값을 입력 후 “Execute” 버튼을 클릭합니다.

fastapi pg 25

API를 실행 후 “Response body” 에서 결과를 확인 합니다.

fastapi pg 26

이상으로 FastAPI의 기본 API를 구현 해봤습니다.

Written by@[Mr. groove]
There is a difference between knowing the path and walking the path.

GitHub