Published on
•
👁️

FastAPI 설치 및 사용

Authors
  • avatar
    Name
    River
    Twitter

1. FastAPI 시작하기



python 가상 환경

  • 가상 환경 만들기
    $ python -m venv .venv
    • 현재 위치에 .venv 라는 디렉토리가 생긴다.


  • 가상 환경 활성화 / 비활성화
    PS C:\Users\river\git\fastapi-postgre> .venv/Scripts/activate

(.venv) PS C:\Users\river\git\fastapi-postgre> pip list
Package Version
------- -------
pip     24.2

    (.venv) PS C:\Users\river\git\fastapi-postgre> .venv/Scripts/deactivate
    • 활성화를 해줘야 적용이 된다.


  • Python에서 가상환경을 사용하는 이유
    • Java 기반 Spring Boot를 사용할 때와 다르게 Python은 기본적으로 시스템 전역에 패키지를 설치한다.
    • 여러 프로젝트를 작업하는 경우, 각 프로젝트 마다 다른 버전의 패키지를 사용해야 한다면 문제가 생긴다.
    • 각 프로젝트 마다 독립적인 환경을 제공하는 것



fastapi 설치

  • fastapi 모듈 설치하기
    (.venv) PS C:\Users\river\git\fastapi-postgre> pip install fastapi
...

(.venv) PS C:\Users\river\git\fastapi-postgre> pip list
Package           Version
----------------- --------
annotated-types   0.7.0
anyio             4.9.0
fastapi           0.115.12
idna              3.10
pip               24.2
pydantic          2.11.4
pydantic_core     2.33.2
sniffio           1.3.1
starlette         0.46.2
typing_extensions 4.13.2
typing-inspection 0.4.0


  • uvicorn 설치하기
    (.venv) PS C:\Users\river\git\fastapi-postgre> pip install uvicorn
...

(.venv) PS C:\Users\river\git\fastapi-postgre> pip list
Package           Version
----------------- --------
annotated-types   0.7.0
anyio             4.9.0
click             8.1.8
colorama          0.4.6
fastapi           0.115.12
h11               0.16.0
idna              3.10
pip               24.2
pydantic          2.11.4
pydantic_core     2.33.2
sniffio           1.3.1
starlette         0.46.2
typing_extensions 4.13.2
typing-inspection 0.4.0
uvicorn           0.34.2
(.venv) PS C:\Users\river\git\fastapi-postgre>


  • 패키지 설정 저장
    (.venv) PS C:\Users\river\git\fastapi-postgre> pip freeze > requirements.txt

(.venv) PS C:\Users\river\git\fastapi-postgre> code requirements.txt

    annotated-types==0.7.0
anyio==4.9.0
click==8.1.8
colorama==0.4.6
fastapi==0.115.12
h11==0.16.0
idna==3.10
pydantic==2.11.4
pydantic_core==2.33.2
sniffio==1.3.1
starlette==0.46.2
typing-inspection==0.4.0
typing_extensions==4.13.2
uvicorn==0.34.2
    • 이렇게 하는 이유는
      • 사용하는 의존성을 문서화 및 버전 관리
      • 환경 복제
      • 배포 준비
    • requirements.txt는 Spring Boot의 build.gradle 파일과 유사한 역할을 하는 것이다.



간단한 API 서버 생성하기

  1. 루트 디렉토리에 main.py 생성

    from fastapi import FastAPI
from typing import Union

app = FastAPI()

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

@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id}

  2. (.venv) PS C:\Users\river\git\fastapi-postgre> uvicorn main:app --reload

    • -- reload : 소스 파일 수정 시 다시 로딩해주는 옵션

  3. 루트 디렉토리에 run.bat이라는 파일 생성

    uvicorn main:app --reload
    • 간단하게 입력하기 위해서
      C:\Users\river\git\fastapi-postgre>uvicorn main:app --reload
INFO:     Will watch for changes in these directories: ['C:\\Users\\river\\git\\fastapi-postgre']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [40632] using StatReload
INFO:     Started server process [119752]
INFO:     Waiting for application startup.
INFO:     Application startup complete.



API 문서화 : Swagger UI



API 문서화 : ReDoc





2. 컨트롤러 확장하기

  • 기존 main.py

    from fastapi import FastAPI
from typing import Union

app = FastAPI()

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

@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id}


  • 따로 API를 컨트롤러(router)로 분리하기 : items.py
    # controller/items.py

from typing import Union
from fastapi import APIRouter

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

@router.get("/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}
    1. APIRouter를 import 한다.
    2. router를 만들어준다.
      • prefix 설정은 Spring Boot의 @RequestMapping("/api/v1/animals")과 같다.
      • tags 설정은
      • responses 설정은
    3. 기존의 @app.get("/items/{item_id}")를 @router.get("/{item_id}")로 전환


  • main.py에서 items.py를 추가하여 사용한다.
    from fastapi import FastAPI
from controller import items

app = FastAPI()

app.include_router(items.router)

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



새로운 컨트롤러 추가하기

  • users.py 생성
    # controller/users.py

from fastapi import APIRouter

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

@router.get("/{user_id}")
def read_user(user_id: int):
    return {"user_id": user_id}
    • from fastapi import APIRouter
      • APIRouter는 FastAPI에서 라우터를 구성하기 위한 모듈화 도구
      • Spring Boot의 @RestController
    • router = APIRouter(...)
      • 실제로 API 경로를 등록할 "라우터"
      • prefix="/users"
        • Spring Boot의 @RequestMapping("/users")와 동일
      • tags=["users"]
        • OpenAPI(Swagger) 문서에서 이 라우터를 "users" 그룹으로 묶기
      • responses={404: {"description": "Not Found"}}
        • 자동 문서화 시, 404 상태코드에 대한 설명을 미리 지정하는 것
        • 단순 설명하는 용도
    • @router.get("/{user_id}")
      • Spring Boot의 @GetMapping("/{userId}")


  • main.py에 users.py를 추가하기
    from fastapi import FastAPI
from controller import items, users

app = FastAPI()

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

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




3. DB(PostgreSQL) 붙이기



  • psycodg 설치하기
    • PostgreSQL용 Python DB 드라이버 (Java의 JDBC 드라이버)
      • Spring Boot + JPA + MySQL이라면
        implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
runtimeOnly 'com.mysql:mysql-connector-j'  // JDBC 드라이버
    • SQLAlchemy는 ORM만 제공하고, 실제 DB 연결은 하지 않기 때문에 psycodg가 필요하다.
      (.venv) PS C:\Users\river\git\fastapi-postgre> pip install "psycopg[binary,pool]"

...

(.venv) PS C:\Users\river\git\fastapi-postgre> pip list
Package           Version
----------------- --------
annotated-types   0.7.0
anyio             4.9.0
click             8.1.8
colorama          0.4.6
fastapi           0.115.12
h11               0.16.0
idna              3.10
pip               24.2
psycopg           3.2.8
psycopg-binary    3.2.8
psycopg-pool      3.2.6
pydantic          2.11.4
pydantic_core     2.33.2
sniffio           1.3.1
starlette         0.46.2
typing_extensions 4.13.2
typing-inspection 0.4.0
tzdata            2025.2
uvicorn           0.34.2


  • DB 연결을 위한 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
    • Spring Boot + JPA 사용 시, 기본적으로 HikariCP 커넥션 풀이 자동으로 설정
    • 그와 달리 FastAPI에서는 Connection Pool을 직접 설정을 해줘야 성능 저하가 없다.


  • 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
    • import psycopg
      • psycopg3 라이브러리의 핵심 모듈로, PostgreSQL 드라이버로 직접 연결 및 쿼리 실행할 때 사용
    • import psycopg_pool
      • DB 커넥션을 매번 새로 만들지 않고, 재사용을 통해 성능을 향상시키는 모듈 (고속 커넥션 풀 모듈)
      • psycopg3에서 제공하는 경량 커넥션 풀(Pool)
    • with pool_default.connection() as conn:
      • psycopg_pool.ConnectionPool에서 Connection 하나 획득하여 conn이라고 명명
      • with 문을 사용하여 Connection 자동 반납 (java의 try-with-resources 문)
    • cur = conn.cursor(row_factory=psycopg.rows.dict_row)
      • 커서를 생성
        • 커서(cursor)란 DB Connection을 통해 쿼리를 실행할 수 있게 해주는 객체
        • Java의 경우 PreparedStatement, ResultSet
        • cursor.execute("SQL")은 PreparedStatement.executeQuery()와 매우 유사
      • row_factory=psycopg.rows.dict_row는 row_factory를 dict_row로 설정하여 반환 결과를 딕셔너리(dict) 형태로 바꿔주는 설정
        • 컬럼명이 key 값이다.
        • 딕셔너리(dict) 자료형은 Java의 Map<key, value>와 매우 유사한 자료 구조
    • cur.execute("SQL문").fetchall()
      • SQL 문장을 실행해서 결과를 전부 가져온다.
      • fetchall()은 결과 전체를 리스트 형태로 가져오는 메서드
        • fetchone()
        • fetchmany(n) (단, 일부)
    • 예외 처리
      • OperationalError - 연결 문제나 DB 서버 장애 등
      • ProgrammingError - SQL 문법 오류 등
      • IntegrityError - 제약 조건 위반 (PK 중복, NULL 제한 등)


  • 새로운 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 Fount"}},
)

@router.get("/list")
def list_admin():
    results = pgsql_test.list_admin()
    return results
    • FastAPI는 내부적으로 dict를 자동으로 JSON 응답으로 직렬화(serialize) 해준다.
      • FastAPI는 함수에서 dict, list, Pydantic 모델 등을 반환하면 자동으로 JSON으로 직렬화해서 응답
    • 그냥 dict만 return 하면 된다.


  • main.py에 admins.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
      [
  {
    "admin_no": 1,
    "login_id": "honggildong",
    "passwd": "1234",
    "nick": "HONG",
    "email": "hgd@gmail.com"
  },
  {
    "admin_no": 2,
    "login_id": "jangnara",
    "passwd": "1234",
    "nick": "JANG",
    "email": "jnr@gmail.com"
  }
]




4. stored procedure 사용하기

  • Stored Procedure(저장 프로시저)는 데이터베이스에 미리 저장된 SQL 코드 블록(함수)으로 직접 SQL을 보내는 대신 로직을 실행하게 하는 것


  • 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:
            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

# 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
    • cur.execute("CALL SP_L_ADMIN('out1')") results = cur.execute("FETCH ALL FROM out1").fetchall() conn.commit()
    • 동일 결과 출력한다.
Discuss on Twitter • View on GitHub