Flask 구조 : DB 설정

DB 설정 목표

1. 내가 원하는 모양대로 테이블이 설정 되어야한다.
2. test, dev, prod 등 각각 다른 DB 환경에서 같은 테이블로 작동되어야한다.

기본 구조

application
|src
|──domain # http 관련 
|────|main
|────────|__init__.py
│──database.py # db 관련
|──model # orm 모델 관련
|────|models.py # 
|──__init__.py # flask app 관련
|──config.py # 각종 설정
|main.py # 실제 서버를 띄우는 역할

DB모듈 설치 : Flask_migrate

• alembic 기반의 Flask용으로 만들어진 모듈
• pip install flask_migrate

database.py 작성

• database 관련 모듈들을 모두 모아두는 곳

#src/database.py
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import Session
from flask_migrate import Migrate

migrate = Migrate()
db = SQLAlchemy()

• from src.database import db, migrate 등등으로 사용할 수 있다.
• flask_migrate를 사용하기 위해선 model 들이 db.Model을 상속 받아야한다.
• 이 부분이 sqlalchemy랑 다른데, sqlalchemy의 경우 Model들이 Base를 상속받는다.

#src/model/models.py
from src.database import db


class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128))

마이그레이션

• 마이그레이션 명령어는 flask db init 이다.
• FLASK_APP이 없다고 에러가 나오는 경우 환경변수로 연결해줘야한다
• $export FLASK_APP={app이 실행될 py}
• 예) $export FLASK_APP=main.py

# main.py
from src import create_app
app = create_app("dev")

if __name__ == '__main__':
	app.run()

• Flask가 db를 인식해줄 수 있게 연결해줘야한다.

# __init__.py
from flask import Flask
from src.database import db, migrate
from src.config import config

def create_app(env):
	app = Flask(__name__)
    app.config.from_object(config[env])
    db.init_app(app)
    migrate.init_app(app, db)
    return app

• 마이그레이션을 하게 되면 env에 맞는 config를 찾아서 해당 환경에 적혀져 있는 DB주소로 마이그레이션이 진행된다.
• 이를 위해 config는 구분되어야한다.

import os

basedir = os.path.abspath(os.path.dirname(__file__))


class Config(object):
    DEBUG = False
    TESTING = False
    CSRF_ENABLED = True
    SECRET_KEY = "super-secret"
    SQLALCHEMY_DATABASE_URI = (
        os.environ.get("DEV_DATABASE_URL")
        or "driver://주소"
    )


class ProductionConfig(Config):
    DEBUG = False
    NAME = "PROD"
    SQLALCHEMY_DATABASE_URI = (
        os.environ.get("DEV_DATABASE_URL")
        or "driver://주소")


class DevelopmentConfig(Config):
    DEVELOPMENT = True
    DEBUG = True
    NAME = "DEV"
    SQLALCHEMY_DATABASE_URI = (
        os.environ.get("DEV_DATABASE_URL")
        or "driver://주소")


class TestingConfig(Config):
    TESTING = True
    NAME = "TEST"
    SQLALCHEMY_DATABASE_URI = (
        os.environ.get("DEV_DATABASE_URL")
        or "driver://주소")


config = {"test": TestingConfig, "dev": DevelopmentConfig, "prod": ProductionConfig}

• $flask db init을 하게 되면 기본적으로 migration 폴더가 생성된다.
• 이후 flask db migrate를 하면 위에서 작성한 model의 스키마 정보를 읽어서 마이그레이션을 진행한다.
  
• 문제는 위 처럼 no detected인데, 이 경우는 우리가 작성한 model을 import 하지 않았기 때문이다.

# src/domain/main/__init__.py
from flask import Flask, Blueprint

from src.model.models import User

api_main = Blueprint("main", __name__)


@api_main.route("/")
def index():
    return "HELLO_WORLD"

• 위 처럼 model User를 import 하고 다시 마이그레이션을 진행하면
  
  detected가 된다.
• 이후 flask db upgrade를 하게 되면 table이 생성된다.
• FLASK_APP으로 설정된 main.py의 환경을 따라가기 때문에 'dev' 환경으로 마이그레이션이 진행되고 DevConfig에 있는 DB 환경으로 테이블이 생성된다.
  

테스트 마이그레이션

• 테스트의 경우 test 환경, 즉 test용 DB에 마이그레이션이 진행되어야한다.
• 이를 위해선 create_app을 할때 test 환경을 건네주어야한다.

#conftest.py

@pytest.fixture(scope="session")
def app():

    app = create_app("test")
    app_context = app.app_context()
    app_context.push()

    yield app

    app_context.pop()

• db를 생성해줘야한다.

# conftest.py
@pytest.fixture(scope="session")
def db(app):
    with app.app_context():
        _db.create_all()
        yield _db
        _db.drop_all()

• 위까지 진행이 되면 test db로서 테이블이 생성까지만 된다.
• 테스트시에 해당 DB를 사용하기 위해서, 즉 session을 사용하기 위해선 위 test db로 만들어진 session을 등록(?) 해줘야한다.
• session을 사용하는 경우는 2가지이다.

1. 실제 app 안에서 session을 사용하는 경우

from src.database import db
@app.route("/")
def main():
	db.session.add(...)
    db.session.commit()

2. test_code안에서 사용하는 경우

def test_is_session_works(session):
	session.add(...)
	session.commit()

• test 환경에서 위와 같은 2가지 상황을 모두 만들어주기 위해선 아래와 같은 코드를 작성해야한다.

# conftest.py
@pytest.fixture(scope="function")
def session(app, db, request):
    """Creates a new database session for each test, rolling back changes afterwards"""
    connection = _db.engine.connect()
    transaction = connection.begin()

    options = dict(bind=connection, binds={})
    session = _db.create_scoped_session(options=options)

    _db.session = session # 1번 상황을 위한 세팅

    yield session # 2번 상황을 위한 세팅

    transaction.rollback()
    connection.close()
    session.remove()

• 이렇게 되면 def test_ 에서는 session을 통해서 직접 접근이 가능하고 app_context 안에선 db.session으로 접근이 가능하다.