Flask – SQLAlchemy 설치 및 사용법

설치 및 설정

• 관련 패키지 설치
  • flask-migrate – Flask 웹 프레임워크에서 ORM을 사용할 수 있도록 해주는 모듈
  • pymysql – MySQL을 Python에서 사용할 수 있도록 해주는 모듈

pip install flask-migrate
pip install pymysql

# 추가로 freeze
pip freeze > requirements.txt

• 초기 선언 및 설정

from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

db = SQLAlchemy()
migrate = Migrate()

db.init_app(app)
    if str(app.config['SQLALCHEMY_DATABASE_URI']).startswith('sqllite'):
        migrate.init_app(app, db, render_as_batch=True)
    else:
        migrate.init_app(app, db)

# 환경설정
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://root:password@localhost/database?charset=utf8' # 프로그램://계정:비밀번호@호스트:포트/데이터베이스명

# sqllite의 경우 Alter문에 대해 모듈이 지원하지않음
# 그러므로 MetaData 클래스를 사용하여 규칙을 정의해야하므로 설정시 옵션 render_as_batch=True을 추가해야함

ERROR [root] Error: No support for ALTER of constraints in SQLite dialectPlease refer to the batch mode feature which allows for SQLite migrations using a copy-and-move strategy.

---

flask DB 관련 주요 Command 정리 (flask db –help) 및 관련 에러

명령	의미
branches	현재 브랜치의 지점을 보여줍니다
current	각 DB의 현재 형상을 보여줍니다
downgrade	DB를 이전 버전으로 변경
upgrade	DB를 최신 버전으로 변경
edit	형상 파일 수정
heads	Show current available heads in the script directory
history	변경내역 조회
init	새로운 마이그레이션 생성
list-templates	List available templates
merge	두 형상 파일 병합
migrate	형상 파일 자동 생성
revision	형상 파일 생성
show	Show the revision denoted by the given symbol
stamp	‘stamp’ the revision table with the given revision

# 마이그레이션 repository 생성
flask db init
이 과정에서 생성된 테이블 (alembic_version)은 현재 자신의 db 버전을 가리키기 위해 생성되는 테이블

# 현재 DB 형상 확인
flask db current

# 현재 형상 마이그레이션 생성
# 메시지는 " 쌍따옴표를 쓰도록하자 '은 에러나는 경우있음
flask db migrate -m "Init database"

# 바뀐 DB 형상 내용을 DB에 적용
flask db upgrade

# 어떠한 오류로 upgrade와 migrate가 안될때
# ex) Cannot Add Foreign Key Constraint 에러 발생
# 잘못된 칼럼 타입을 넣고 foreign key를 생성하려하니 upgrade가 안되는 상황
# 변경하려해도 이미 잘못된 타입을 migrate한 상태라 upgrade가 안되고
# downgrade 후 migrate 진행시 [Target database is not up to date.] 에러로 역시 안될때
flask db stamp head
# head에서 다시 migrate
flask db migrate

# pymysql 모듈을 추가했는지 확인해보자
ModuleNotFoundError: No module named 'MySQLdb'

# mysql이 동작하는지 연결정보는 확인해보자
pymysql.err.OperationalError: (2003, "Can't connect to MySQL server on 'localhost' ([WinError 10061] 대상 컴퓨터에서 연결을 거부했으므
로 연결하지 못했습니다)")

---

flask_sqlalchemy에서 자주 사용하는 속성

• Table
  • 관리할 테이블 객체 생성

table_name = db.Table(
    'table_name', # 테이블명

    db.Column('column_name1', db.Integer),
    db.Column('column_name2', db.String(100)),
    db.Column('column_name3', db.Text),
    db.Column('column_name4', db.Boolean),
    db.Column('column_name5', db.DateTime)
)

• Model
  • 관리할 테이블을 Model 객체로 지정

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class TABLE_A(db.Model):
     pass

• __tablename__
  • 테이블명을 멤버 변수로 지정

• __table_args__
  • 테이블 내 속성을 정의
• UniqueConstraint
  • 복수 칼럼 병합 Unique 설정

# 방법1
__table_args__ = {'mysql_collate': 'utf8_general_ci'}

# 방법2
__table_args__ = (
    db.UniqueConstraint(
        'column_a',
        'column_b',
        name="uk_a_b"
    ), # "," 안붙이면 에러 발생 (why..)
)

• Column
  • 테이블의 속성을 정의

user_id = db.Column(
    db.Integer,
    db.ForeignKey(
        'user.id',
        ondelete='CASCADE'
    ),
    nullable=False
)

• relationship
  • 테이블 간의 관계를 정의

# model a 내부에 선언된 상태
table_b = db.relationship(
    'model_name', # 연관된 모델 지정
    secondary=tabel_name, # 다대다 관계 테이블을 참조
    # model a에서 b를 참조하듯 b에서도 a를 참조할 수 있도록 선언
    backref=db.backref('table_a') 
)

# 사용 예시
from [model_path] import Model_A
from [model_path] import Model_B

Model_A.query.first().table_b
Model_B.query.first().table_a

• 추가) DB 테이블 모델화 예시

from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

db = SQLAlchemy()

class User(db.Model):
    __tablename__ = 'user'
    __table_args__ = {'mysql_collate': 'utf8_general_ci'}

    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.String(20), unique=True, nullable=False)
    user_name = db.Column(db.String(20), nullable=False)
    password = db.Column(db.String(100), nullable=False)
    joindate = db.Column(db.DateTime, default=datetime.utcnow())

    def __init__(self, user_id, user_name, password) -> None:
        self.user_id = user_id
        self.user_name = user_name
        self.password = password

    @classmethod
    def find_one_by_user_id(cls, user_id):
        return User.query.filter_by(user_id=user_id).first()

    def as_dict(self):
        return {x.name: getattr(self, x.name) for x in self.__table__.columns}

---

쿼리 (Query) 사용법

from app_name import db
from app_name.models.user import User
from app_name.models.payment import Payment

• Select – 검색

# 쿼리 확인
print(User.query)

# select * from user;
User.query.all()

# select * from user limit 0, 1;
User.query.first()

# primary key 값 검색
# select * from user where id = 3;
User.query.get(3)

# 검색 내용이 없을 경우 404 처리
User.query.get_or_404(3)

# LIMIT절
# select * from user limit 0, 2;
User.query.limit(2).all()

# Where절 + Like절 = select * from user where user_name like 'jpro%';
User.query.filter(User.user_name.like('jpro%')).all()

• ORDER BY – 정렬

User.query.filter(User.user_name.like('jpro%')).order_by(User.joindate.desc(), User.id).all()

• GROUP BY – 그룹화

User.query.query.group_by(User.user_id).all()

• JOIN – 조인

# Join
db.session.query(User, Payment).filter(User.id == Payment.user_id).all() 

# Left Join
db.session.query(User).\
    outerjoin(Payment, User.id == Payment.user_id).\
    all()

• IN, NOT IN – 포함과 제외

# IN
User.query.filter(User.user_name.in_(('jpro', 'kpro'))).all() 

# NOT IN
User.query.filter(~User.user_name.in_(('jpro', 'kpro'))).all() 

• EXISTS ANY, HAS
  • any() – 다대다 관계에서 하나라도 포함
  • has() – 다대일 관계에서 하나라도 포함

filter(User.addresses.any(Address.email_address.like('%gmail%')))

filter(~Address.user.has(User.name=='jack')).all()

• paginate
  • 페이지 관리용 함수
  • 결과값 호출 (item) -> pages.item

pages = data.order_by(
    MemoModel.created_at.desc()
).paginate(
    page=page,
    per_page=10
)

• Update – 값 변경

# 특정 값 변경
user = User.query.get(7)
user.user_name = 'jpro'

# 범위 변경
User.query.filter(User.id > 1).update({
    'user_name':'jpro7'
})

• Insert – 값 저장

db.session.add(
    User(
        user_id="jpro10@jpro.blog", 
        user_name="jpro", 
        password="1234"
    )
)

• Delete – 값 삭제

# 모두 삭제
User.query.delete()

# 특정 값 삭제
user = User.query.first()
db.session.delete(user)

• Commit, Rollback, Flush
  • Commit – 트랜잭션 적용
  • Rollback – 트랜잭션 복원
  • Flush – 트랜잭션 전송 (데이터베이스로 변경 사항을 전송하지만 Commit이 완료되지 않은 상태

db.session.commit()
db.session.rollback()
db.session.flush()

---

참고 자료

• Flask-SQLAlchemy
• Flask-Migrate
• SQLAlchemy-Migrate
• SQLAlchemy – API
• SQLAlchemy – SELECT, INSERT, DELETE
• SQLAlchemy – JOIN