FastAPI (3) : DB(PostgreSQL) 붙여보기

 

 

 

이번에는 FastAPI 프레임워크에 PostgreSQL 을 붙여서 동작시켜보자. 우선, scoop 을 이용해서 PostgreSQL을 설치해준다. scoop install postgresql 을 실행한다.

pg_ctl start 를 입력해서 PostgreSQL을 기동합니다.

psql -U postgres 를 입력해서 PostgreSQL에 관리자로 로그인합니다.

다음처럼 새로운 사용자 계정을 생성합니다. 사용자 이름과 비밀번호는 원하는 것으로 설정하면 되겠습니다.

그리고, Database 를 생성합니다. 데이터베이스 이름은 test_db 로 하였고, 소유주는 test_user 로 하였습니다.

\q 로 PostgreSQL에서 빠져나와서 새로 새성한 사용자로 로그인해봅니다.

새로운 테스트용 테이블도 생성해봅니다.

CREATE TABLE TB_ADMIN
(
    ADMIN_NO Serial NOT NULL,
    LOGIN_ID Varchar(20) NOT NULL UNIQUE,
    PASSWD Varchar(20) NOT NULL,
    NICK Varchar(20) NOT NULL,
    EMAIL Varchar(40),
    PRIMARY KEY (ADMIN_NO)
) Without Oids;

테스트용 데이터도 넣어봅시다.

INSERT INTO TB_ADMIN(LOGIN_ID, PASSWD, NICK, EMAIL)
VALUES('honggildong', 'ajtwlddl', 'HONG', 'hgd@gmail.com');
 
INSERT INTO TB_ADMIN(LOGIN_ID, PASSWD, NICK, EMAIL)
VALUES('jangnara', 'dlQmsdl', 'JANG', 'jnr@gmail.com');

데이터가 잘 들어갔는지 확인해봅니다.

PROCEDURE 와 FUNCTION을 각각 만들어보겠습니다.

CREATE OR REPLACE PROCEDURE public.SP_L_ADMIN(out1 refcursor)
	LANGUAGE plpgsql
AS $procedure$
	BEGIN
		OPEN out1 FOR
		SELECT ADMIN_NO, LOGIN_ID, PASSWD, NICK, EMAIL FROM TB_ADMIN;
	END;
$procedure$
;

 

CREATE OR REPLACE FUNCTION public.FN_L_ADMIN(out1 refcursor)
	RETURNS SETOF refcursor
	LANGUAGE plpgsql
AS $function$
	BEGIN
		OPEN out1 FOR
		SELECT ADMIN_NO, LOGIN_ID, PASSWD, NICK, EMAIL FROM TB_ADMIN;
		RETURN NEXT out1;
	END;
$function$
;

그리고, SP_L_ADMIN 와 FN_L_ADMIN을 호출해서 TB_ADMIN 의 내용을 조회해 보겠습니다.

여기까지 해서, Database 쪽은 준비가 되었습니다. 이제, FastAPI 에서 DB 연결부터 조회하는 것까지 해봅시다. ORM 같은 것은 사용하지 않고, 쌩 SQL과 Stored Procedure 를 이용하는 법을 알아봅니다.

먼저 pip install "psycopg[binary,pool]" 을 실행해서 PostgreSQL 패키지를 설치합니다.

 

fapi 폴더 아래에 config 폴더를 생성하고, 그 안에 config.py 를 만들어줍니다.

# config/config.py

PGSQL_TEST_DATABASE_STRING = "host=127.0.0.1 dbname=test_db user=test_user password=test123 port=5432"
PGSQL_TEST_POOL_MIN_SIZE = 10
PGSQL_TEST_POOL_MAX_SIZE = 10
PGSQL_TEST_POOL_MAX_IDLE = 60

fapi 폴더 아래에 model 폴더를 생성하고, 그 안에 pgsql_test.py 를 만들어줍니다.

# model/pgsql_test.py

import psycopg
import psycopg_pool
from config import config

pool_default = psycopg_pool.ConnectionPool(config.PGSQL_TEST_DATABASE_STRING,
                                            min_size=config.PGSQL_TEST_POOL_MIN_SIZE,
                                            max_size=config.PGSQL_TEST_POOL_MAX_SIZE,
                                            max_idle=config.PGSQL_TEST_POOL_MAX_IDLE)

def list_admin():
    with pool_default.connection() as conn:
        cur = conn.cursor(row_factory=psycopg.rows.dict_row)

        try:
            results = cur.execute("SELECT * FROM TB_ADMIN").fetchall()
        except psycopg.OperationalError as err:
            print(f'Error querying: {err}')
        except psycopg.ProgrammingError as err:
            print('Database error via psycopg.  %s', err)
            results = False
        except psycopg.IntegrityError as err:
            print('PostgreSQL integrity error via psycopg.  %s', err)
            results = False
    return results

이제, controller 폴더 아래에 admins.py 를 만들자.

# controller/admins.py

from fastapi import APIRouter
from model import pgsql_test

router = APIRouter(
    prefix="/admins",
    tags=["admins"],
    responses={404: {"description": "Not found"}},
)

@router.get("/list")
def list_admin():
    results = pgsql_test.list_admin()
    return results

main.py 에도 admins 를 추가해준다.

# main.py

from fastapi import FastAPI
from controller import items, users, admins

app = FastAPI()

app.include_router(items.router)
app.include_router(users.router)
app.include_router(admins.router)

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

웹브라우저에서 http://localhost:8000/admins/list 를 호출하자.

이번에는 pgsql_test.py 의 list_admin() 함수를 아래처럼 바꾸어본다. 위에서는 일반 SQL을 실행했다면, 이번에는 Stored Procedure 를 실행해보는 것이다.

def list_admin():
    with pool_default.connection() as conn:
        cur = conn.cursor(row_factory=psycopg.rows.dict_row)

        try:
            cur.execute("CALL SP_L_ADMIN('out1')")
            results = cur.execute("FETCH ALL FROM out1").fetchall()
            conn.commit()
        except psycopg.OperationalError as err:
            print(f'Error querying: {err}')
        except psycopg.ProgrammingError as err:
            print('Database error via psycopg.  %s', err)
            results = False
        except psycopg.IntegrityError as err:
            print('PostgreSQL integrity error via psycopg.  %s', err)
            results = False
    return results

그리고, 또 다시 http://localhost:8000/admins/list 를 호출해본다.