Python - FastAPI DB Connection ( SQLAlchemy )

필수 설치 요소

pip3 install sqlalchemy
# ORM을 통하여 DB 쿼리문을 작성하기 위해 설치

pip3 install python-dotenv
# DB관련 정보를 입력할 때, 환경변수를 통하여 내용을 입력하기 위해 dotenv 설치

 

폴더구조

└── app
    ├── __init__.py
    ├── apis
    │   ├── __init__.py
    │   └── test.py
    ├── core
    │   ├── __init__.py
    │   └── config.py
    ├── crud
    │   ├── __init__.py
    │   └── crud_test.py
    ├── db
    │   ├── __init__.py
    │   ├── connection.py
    │   ├── models
    │   │   ├── __init__.py
    │   │   └── test_model.py
    │   └── session.py
    ├── main.py
    └── routes
        ├── __init__.py
        └── test.py

• apis : 메인 로직을 작성하는 부분( MVC 패턴의 Controller와 같은 역할 )
• core : core의 config.py에는 DB URL을 반환하는 클래스가 포함되어있다
• crud : ORM 형식으로 DB 쿼리문이 작성된 파일들이 모여있는 폴더
• db : DB 연결, DB 세션 관리, 모델들을 정의한 폴더
• routes : URL 경로에 따라 api 함수들을 나눠주는 역할
• main.py : FastAPI를 실행하고 routes폴더들의 다른 라우트들을 불러와 포함시키는 역할
• __init__.py : 해당 폴더가 패키지의 일부라는 것을 나타낸다. 해당 파일에 작성되어야 할 내용은 없다. Python 3.3버전 이후에는 해당 파일이 존재하지 않아도 패키지의 일부로 인식할 수 있지만 하위 버전의 Python과 호환을 위해서 만들어두는 것을 권장한다.

 

파일 설정 방법

1. main.py

# main.py

import os
from typing import Optional
from fastapi import FastAPI
from routes.test import router as test_router

app = FastAPI() # FastAPI 모듈
app.include_router(test_router) # 다른 route파일들을 불러와 포함시킴

@app.get("/") # Route Path
def index():
    return {
        "Python": "Framework",
    }

include_router를 통해 다른 라우트들을 포함시켜 가져올 수 있다.

 

 

2. routes/test.py

from typing import Optional
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from db.connection import get_db
from crud import crud_test
from apis import test # main logic

router = APIRouter(
    prefix="/items", # url 앞에 고정적으로 붙는 경로추가
) # Route 분리

@router.get("/test_route") # Route Path
def test_index(db: Session = Depends(get_db)):
	
	res = test.test_index(db=db) # apis 호출
	
	return {
        "res" : res,
	} # 결과

FastAPI의 Depends 함수를 통해 API가 호출되어 DB 세션을 시작하여 사용할 수 있도록 해준다.

 

 

3. db/connection.py

from db.session import SessionLocal

def get_db():
	db = SessionLocal()
	try:
		yield db # DB 연결 성공한 경우, DB 세션 시작
	finally:
		db.close()
		# db 세션이 시작된 후, API 호출이 마무리되면 DB 세션을 닫아준다.

 

 

4. db/session.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base # Base 생성
from core.config import settings


SQLALCHEMY_DATABASE_URL = settings.DATABASE_URL
engine = create_engine(SQLALCHEMY_DATABASE_URL)

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

• create_engine :  인자값으로 DB URL을 추가하면 DB Host에 DB 연결을 생성한다. 이 함수가 DB연결의 출발점이다.
• sessionmaker : 호출되었을 때, 세션을 생성해준다.
  • autocommit : api가 호출되어 DB의 내용이 변경된 경우, 자동으로 commit하며 변경할지에 대한 여부를 결정한다. False로 지정한 경우에는, insert, update, delete 등으로 내용이 변경됬을 때, 수동적으로 commit을 진행해주어야 한다.
  • autoflush : 호출되면서 commit되지 않은 부분의 내역을 삭제할지의 여부를 정하는 부분이다.
  • bind : 어떤 엔진을 통해 DB연결을 할지 결정하는 부분이다. MySQL, PostgreSQL 등 여러 SQL의 DB URL 중 어느 SQL제품으로 연결을 진행할지 선택하는 부분이다. 위의 부분에서는 engine변수가 하나밖에 선언되어있지 않지만, SQL을 여러 종류 쓰는 경우, 각 SQL에 맞게 해당 부분이 여러종류로 나뉠 수 있다.
• delarative_base() : 상속된 DB모델 클래스들을 자동적으로 연결시켜주는 역할을 한다. 쉽게 말해, 테이블명이 일치하는 모델을 찾아 쿼리문을 실행시켜준다.

 

5. core/config.py

# core/config.py

import os
from dotenv import load_dotenv

load_dotenv()

class Settings:
	
    DB_USERNAME : str = os.getenv("DB_USERNAME")
    DB_PASSWORD = os.getenv("DB_PASSWORD")
    DB_HOST : str = os.getenv("DB_HOST","localhost")
    DB_PORT : str = os.getenv("DB_PORT",3306)
    DB_DATABASE : str = os.getenv("DB_DATABASE")
	
    DATABASE_URL = f"mysql+pymysql://{DB_USERNAME}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_DATABASE}"

settings = Settings()

.env파일에 저장된 DB 관련 내용을 바탕으로 DB URL 문자열을 선언한다.

• mysql+pymysql : mariadb를 지원하는 버전일 경우
• mariadb+pymysql : mariadb만 지원하는 버전일 경우
• posgresql : postgreSQL을 사용하는 경우

사용하는 DB에 따라, 위의 내용이 달라지게 된다. 위의 내용이 달라질 경우, create_engine 부분에 들어가는 DB URL이 바뀌게 되는 점을 유의하여야 한다.

 

---

6. routes/test.py

# routes/test.py

@router.get("/test_route") # Route Path
def test_index(db: Session = Depends(get_db)):
	
	res = test.test_index(db=db) # apis 호출
	
	return {
        "res" : res,
	} # 결과

연결을 진행한 후, api를 호출하는 함수를 확인한다. apis/test.py의  test_index함수를 호출한다.

 

 

7. apis/test.py

# apis/test.py

from sqlalchemy.orm import Session
from crud import crud_test

def test_index(db):
    something = crud_test.get_items(db)
    return something

apis폴더의 파이썬 파일들은 MVC패턴의 컨트롤러와 비슷한 개념으로 보면된다. 이 부분에서 주요한 로직을 입력하게 된다. DB와 연결하여 데이터를 불러오거나 변경하는 부분은 crud라는 폴더를 생성하여 DB 내용만 처리하는 부분을 따로 만들어 놓았다.

 

 

8. crud/crud_test.py

# crud/crud_test.py

from sqlalchemy.orm import Session
from db.models.test_model import Test

def get_items(db: Session):
    return db.query(Test).all()

SQLAlchemy ORM을 통해 Test모델의 모든 데이터들을 가져오는 구문을 만들었다. db.query에 Test라는 모델명이 들어간 이유는 해당 모델을 바탕으로 상속된 DB 테이블의 데이터를 가져오거나 변경한다는 것을 의미한다.

 

 

9. models/test_model.py

# models/test_model.py

from sqlalchemy import Column, String, Integer, ForeignKey
from sqlalchemy.orm import relationship
from db.session import Base


class Test(Base):
    __tablename__ = "test"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String)
    description = Column(String)

• __tablename__ : DB테이블명. 테이블명 밑 변수들은 해당 테이블에 속해있는 칼럼들을 뜻한다.

 

uvicorn main:app --reload
# main.py 파일이 존재하는 경로에서 FastAPI 실행

 

api 결과

해당 url을 통해 api를 호출하면 실행결과에 따른 결과값이 반환되는 것을 확인할 수 있다.

 

---

/docs

 

/redoc

/docs와 /redoc에서 API를 실행하거나 목록을 확인할 수 있다.

---

 

https://docs.sqlalchemy.org/en/14/core/type_basics.html

Column and Data Types — SQLAlchemy 1.4 Documentation

SQLAlchemo ORM 모델관련 데이터타입과 기본키, 외래키와 같은 옵션들은 공식홈페이지에 잘 정리되어 있다.