flask with ORM

ORM (Object-Relational Mapping)

 

객체 지향 프로그래밍 언어에서 관계형 데이터베이스의 데이터를 객체로 매핑하는 방법

Flask-SQLAlchemy는 Python 객체와 데이터베이스 테이블 간의 매핑을 자동으로 처리 -> SQL 쿼리 작성 없이 Python 객체를 다루듯 데이터베이스와 상호작용 가능

# ORM 설치
$ pip install flask-migrate

 

ORM 구성

 

1. config.py

• .env 파일: 애플리케이션 외부에서 환경 변수를 관리하는 파일
  • 비밀번호, API 키, 데이터베이스 URI와 같은 민감한 설정 값들이 포함

• SQLALCHEMY_DATABASE_URI: 데이터베이스 URI를 설정
• SQLALCHEMY_TRACK_MODIFICATIONS: False로 설정하여 Flask가 객체 변경 추적을 하지 않게 한다.

import os

BASE_DIR = os.path.dirname(__file__)

SQLALCHEMY_DATABASE_URI = 'sqlite:///{}'.format(os.path.join(BASE_DIR, 'test.db'))
SQLALCHEMY_TRACK_MODIFICATIONS = False

 

 

2. app.py

• create_app(): Flask 애플리케이션을 팩토리 패턴으로 생성
  • db.init_app(app): Flask-SQLAlchemy의 db 객체를 애플리케이션에 연결
  • migrate.init_app(app, db): 데이터베이스 마이그레이션을 처리하는 Flask-Migrate의 초기화 코드

from flask import Flask, jsonify
from models import db, Product, User
import config

app = Flask(__name__)
app.config.from_object(config)

# ORM 초기화
db.init_app(app)

@app.route('/add_product')
def add_product():
    # 새로운 Product 객체를 생성하고 DB에 추가
    new_product = Product(name="Sample Product", price=19.99)
    db.session.add(new_product)
    db.session.commit()
    return "Product added!"

@app.route('/get_products')
def get_products():
    # DB에서 모든 Product 조회
    products = Product.query.all()
    return jsonify([product.to_dict() for product in products])

@app.route('/add_user')
def add_user():
    # 새로운 User 객체를 생성하고 DB에 추가
    new_user = User(username="sample_user", email="user@example.com")
    db.session.add(new_user)
    db.session.commit()
    return "User added!"

@app.route('/get_users')
def get_users():
    # DB에서 모든 User 조회
    users = User.query.all()
    return jsonify([user.to_dict() for user in users])

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

 

 

3. models.py

• db.Model을 상속받은 각 클래스는 SQLAlchemy 모델이 되고, 이 클래스는 데이터베이스 테이블과 매핑된다.
• 각 모델에는 컬럼을 정의
  • ex) Product 모델은 id, name, price 컬럼 / User 모델은 id, username, email 컬럼

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class Product(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)
    price = db.Column(db.Float, nullable=False)

    def to_dict(self):
        return {"id": self.id, "name": self.name, "price": self.price}

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def to_dict(self):
        return {"id": self.id, "username": self.username, "email": self.email}

 

 

4. 마이그레이션

Flask, Django, FastAPI에서 ORM을 사용하려면 마이그레이션을 해야 한다.

모델을 수정하거나 변경할 때  해당 변경을 데이터베이스에 반영하려면 마이그레이션 도구를 사용해서 데이터베이스 스키마를 업데이트

• Flask: Flask-SQLAlchemy와 Flask-Migrate를 사용하여 모델 변경 시 마이그레이션 필요
• Django: 내장된 makemigrations와 migrate 명령어로 자동 마이그레이션
• FastAPI: SQLAlchemy와 Alembic을 사용해 모델 변경 시 마이그레이션 필요

즉, ORM을 사용하면 모델과 데이터베이스 스키마를 일치시키기 위해 마이그레이션을 해야 한다는 것

 

명령어

명령어	설명
flask db init	마이그레이션 환경 초기화 (처음 한 번만 실행)
flask db migrate -m "메시지"	모델 변경 사항을 마이그레이션 파일로 생성
flask db upgrade	마이그레이션을 실행하여 데이터베이스를 최신 상태로 만든다.
flask db downgrade	이전 버전으로 롤백
flask db history	마이그레이션 기록 확인
flask db current	현재 마이그레이션 상태 확인
flask db stamp head	현재 DB 상태를 최신 마이그레이션으로 표시 (실제 업그레이드는 안 함)

 

flask db init
flask db migrate -m "Initial migration"
flask db upgrade

정리해보자

 

ORM 사용 흐름

 

1. 모델 정의
   • Product와 User와 모델을 정의하여 각 객체가 데이터베이스 테이블과 어떻게 매핑되는지 설정
   • 각 모델의 클래스 속성은 데이터베이스 테이블의 컬럼과 대응된다.
2. 데이터베이스 세션 사용
   • db.session.add()와 db.session.commit()을 사용하여 객체를 데이터베이스에 추가하고 저장
     • ex) 새 제품을 데이터베이스에 추가하려면 Product(name='새 제품', price=10.0)와 같은 객체를 만들어 db.session.add()로 추가한 후, db.session.commit()을 호출하여 저장
3. 쿼리 수행
   • Product.query.all() 또는 Product.query.filter_by(name='제품 이름')과 같은 방식으로 SQLAlchemy의 쿼리 API를 사용해 데이터 조회

 

User도 마찬가지 방법으로 DB에 저장되는 것을 확인가능

 

모델 간의 관계 설정

이제 테이블을 연결해보자

Product와 User 모델을 기반으로 Category라는 새로운 모델을 만들어서 두 모델을 서로 연결

 

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class Product(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)
    price = db.Column(db.Float, nullable=False)
    category_id = db.Column(db.Integer, db.ForeignKey('category.id'), nullable=False)  # 외래키로 카테고리와 연결됨
    category = db.relationship('Category', backref=db.backref('products', lazy=True))  # 역참조 설정

    def to_dict(self):
        return {"id": self.id, "name": self.name, "price": self.price}


class Category(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False, unique=True)

    def to_dict(self):
        return {"id": self.id, "name": self.name}

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def to_dict(self):
        return {"id": self.id, "username": self.username, "email": self.email}

• Product 모델에 category_id라는 외래 키를 추가하여 Category 모델과 연결
• category = db.relationship('Category', backref=db.backref('products', lazy=True))는 Product 모델에서 Category 모델을 참조할 수 있게 됨
• Category 모델에서 products를 역참조하면, 특정 카테고리에 속한 모든 제품들을 가져올 수 있다.

 

from flask import Flask, jsonify
from models import db, Product, User, Category
import config

app = Flask(__name__)
app.config.from_object(config)

# ORM 초기화
db.init_app(app)

@app.route('/add_category')
def add_category():
    # 새로운 Category 객체를 생성하고 DB에 추가
    new_category = Category(name="Electronics")
    db.session.add(new_category)
    db.session.commit()
    return "Category added!"

@app.route('/add_product')
def add_product():
    # 새로운 Product 객체를 생성하고 DB에 추가
    # 이미 존재하는 Category ID를 할당하여 Product에 연결
    category = Category.query.first()  # 첫 번째 카테고리 가져오기 (예시로)
    new_product = Product(name="Laptop", price=999.99, category_id=category.id)
    db.session.add(new_product)
    db.session.commit()
    return "Product added!"

@app.route('/get_categories')
def get_categories():
    # DB에서 모든 Category 조회
    categories = Category.query.all()
    return jsonify([category.to_dict() for category in categories])

@app.route('/get_products')
def get_products():
    # DB에서 모든 Product 조회
    products = Product.query.all()
    return jsonify([product.to_dict() for product in products])


@app.route('/get_products_by_category/<int:category_id>')
def get_products_by_category(category_id):
    # 특정 카테고리의 제품들 조회
    category = Category.query.get_or_404(category_id)
    products = category.products  # 역참조를 통해 관련된 제품 조회
    return jsonify([product.to_dict() for product in products])


@app.route('/add_user')
def add_user():
    # 새로운 User 객체를 생성하고 DB에 추가
    new_user = User(username="sample_user", email="user@example.com")
    db.session.add(new_user)
    db.session.commit()
    return "User added!"

@app.route('/get_users')
def get_users():
    # DB에서 모든 User 조회
    users = User.query.all()
    return jsonify([user.to_dict() for user in users])


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

• add_category와 add_product는 새로운 카테고리와 제품을 추가하는 라우트
  • 카테고리를 먼저 추가한 후, 해당 카테고리를 제품에 할당하는 방식
• get_categories는 모든 카테고리를 조회하는 라우트
• get_products_by_category는 특정 카테고리에 속한 제품들을 조회하는 라우트
  • category.products를 통해 해당 카테고리의 모든 제품을 가져오게 됨

 

새로운 모델을 추가했으므로 데이터베이스 마이그레이션을 통해 테이블을 업데이트해야 한다.

 

좀 헷갈리는 부분을 짚고 넘어가자

Blueprint와 Route의 차이

• Route: Flask 애플리케이션의 URL 경로와 함수(핸들러)를 연결하는 기능
• Blueprint: 여러 라우트를 그룹화할 수 있는 방법
  • 큰 애플리케이션에서 라우트를 분리하고 싶을 때 Blueprint를 사용하여 코드를 모듈화