[Python] SQLAlchemy 개념 및 설정 방법

SQLAlchemy 개념 및 사용법에 대해 정리해 보자 ❕

 

Flask로 개발한 적이 있었는데, Flask는 ORM을 내장하지 않아 ORM 라이브러리를 이용해야 했다. 이 때 SQLAlchemy를 사용했는데, 처음 사용해 본 기술이라 다시 한번 정리해보려고 한다. 참고로 Flask에서는 SQLAlchemy를 확장한 Flask-SQLAlchemy도 제공한다.

 

 

먼저 ORM에 대해 정리한 뒤, SQLAlchemy에 대해 알아보자.

 

 

 

1. ORM

---

ORM(Object Relational Mapping)은 객체-관계의 매핑을 의미한다.

ORM은 애플리케이션과 데이터베이스 연결 시 SQL 언어가 아닌 애플리케이션 개발 언어로 데이터베이스에 접근할 수 있게 해주는 도구이다. SQL문법 대신 애플리케이션의 개발언어를 그대로 사용할 수 있게 함으로써 일관성과 가독성을 높여주는 역할을 한다.

 

⇒ 즉, SQL이 아닌 애플리케이션 개발 언어로 데이터베이스를 조작할 수 있게 한다.

 

 

Python에서는 SQLAlchemy, Django ORM, Peewee 등의 ORM이 존재한다. 가장 널리 사용되는 건 SQLAlchemy이고, 장고 프레임워크를 사용한다면 내장된 Django ORM을 사용하면 된다.

 

 

2. SQLAlchemy

---

SQLAlchemy는 Python에서 사용할 수 있는 ORM 중 하나이다. 

 

 

공식문서

SQLAlchemy

 

 

구성

SQLAlchemy는 크게 ORM, Core 두 부분으로 구성되어 있다. ORM뿐만 아니라 SQL 표현 언어인 SQLAlchemy Core도 제공하는데, Core는 ORM보다 더 낮은 수준의 API, 더 세밀한 SQL 쿼리 작성이 가능하다.

 

1. ORM
   • 데이터베이스 테이블을 파이썬 클래스와 매핑하고, 각 테이블 행을 파이썬 객체로 표현
   • 파이썬 코드로부터 적절한 SQL쿼리를 생성해준다.
   • 세션 및 트랜잭션 관리
2. Core
   • 직접 SQL을 작성하고 실행할 수 있는 기능을 제공
   • 데이터베이스와의 연결을 관리하고, SQL 쿼리를 실행한다

 

 

 

3. 설정방법

---

출처 :  https://docs.sqlalchemy.org/en/20/core/engines.html

 

 

3-1. 데이터베이스 연결 설정 - Engine

from sqlalchemy import create_engine

db_api = "pymysql"
db_url = f"mysql+{db_api}://{user}:{password}@{host}:{port}/{database}"
engine = create_engine(url=db_url, pool_size=pool_size, max_overflow=max_overflow, pool_recycle=pool_recycle, pool_timeout=pool_timeout, echo=True)

• Engine은 SQLAlchemy의 Pool과 Dialect 기능을 사용하여 DB API를 사용할 수 있도록 하고 실제 Database에 접근 한다.
• create_engine() : db 연결 함수
  • db_url은 dialect+db_api://~ 형태로 작성한다.
  • db_api를 설정하지 않으면 sqlalchemy의 기본 설정 값이 들어간다. (기본값 참고)
  • pymysql : python으로 mysql에 접근할 수 있는 라이브러리
• 풀링 설정
  • pool_size : 커넥션 수 (default 5)
    • 참고로 0으로 설정하면 하나도 생성되지 않는 것이 아닌 무한대로 생성
  • max_overflow : pool_size 설정 이상으로 연결할 수 있는 커넥션 수 (default 10)
  • pool_recycle : 지정된 시간이 경과하기 전까지 풀에서 커넥션을 재활용 (default -1)
    • 커넥션이 풀에 반환된 후, 지정된 시간 이상동안 유휴 상태면, 해당 커넥션을 닫고 새로운 커넥션을 생성한다.
  • pool_timeout : 커넥션을 얻기까지 기다리는 시간 (default 30s)
  • echo : sql log 여부

 

 

QueuePool 생애주기 (SQLAlchemy 커넥션 풀의 기본은 QueuePool 방식)

1. 큐 풀은 처음부터 미리 커넥션을 만들지 않고, 0개로 시작한다.
2. 요청이 들어왔을 때, 큐 풀에 유효한 커넥션이 없으면 하나 생성한다.
3. 설정된 pool_size까지 만들어진 커넥션을 종료하지 않는다.
4. 요청이 들어왔을 때, 사용할 수 있는 커넥션이 없다면 pool_size를 초과하여 생성한다.
5. pool_size를 초과한 오버플로우 상황에서, `pool_size` + `max_overflow`까지 커넥션을 생성하면서, 오버플로우를 방지하기 위해 새로 들어오는 커넥션을 종료하여 pool_size에 총 커넥션 수를 맞춘다.
6. 큐풀이 관리하는 커넥션이 pool_size + max_overflow까지 다 찬 상황에서 요청이 들어오면, 더이상 커넥션을 생성하지 않고 대기시킨다. (default 30s)
7. 허용된 대기 시간이 초과된다면 TimeoutError 예외가 발생된다.

 

3-2. 세션 관리

from sqlalchemy.orm import scoped_session, sessionmaker

db_session = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))

• session
  
  • 데이터베이스와 클라이언트 사이의 연결을 나타내는 구현체
  • 최초 쿼리 작업을 요청하거나 관련된 객체를 건드리는 순간 세션은 새로운 트랜잭션을 생성하고, 해당 트랜잭션은 세션이 commit 되거나 rollback 되기 전까지 유지된다. 
  • 한 요청에 대해 스레드마다 다른 세션이 생성될 수 있다.
• sessionmaker()
  • db 엔진과 연결한 후 session을 생성하는 역할
• scoped_session()
  • 세션의 생명주기를 관리하고, 설정값에 맞추어 세션을 생성하는 역할
  • scoped_session를 사용한다면 해당 세션이 작업을 마칠 때까지 동일한 세션 객체를 반환하도록 할 수 있다.
  • 따라서 각 스레드나 요청마다 독립적인 세션을 유지하는게 가능해진다.
• 세션의 역할
  • `add()` : 세션이 영속상태가 등록 대기 상태가 되고, commit()을 호출할 때 데이터베이스에 저장되면서 영속 상태가 된다.
  • flush() : 세션의 변경 사항을 데이터베이스에 즉시 반영하지만, 트랜잭션을 commit하지는 않는다.
  • `commit()` : 세션의 변경 사항을 데이터베이스에 반영하고, 트랜잭션을 커밋하여 영구적으로 저장한다.
  • 세션은 영속 상태의 객체에 변경 사항이 발생하면, 이를 감지하고 commit()시 데이터베이스에 반영한다. (dirty checking)

 

 

세션 예시

from sqlalchemy.orm import Session, sessionmaker

session1 = Session(engine)
session2 = Session(engine)

session1 is session2
# False

db_session = sessionmaker(bind=engine)
session1 = db_session()
session2 = db_session()

session1 is session2
# False

일반적으로 `session` 혹은 `sessionmaker`는 독립적인 세션 객체를 만든다. 위 결과처럼 하나의 스레드 안에서 세션을 두 번 호출한다면 서로 다른 세션이 생성되는 것이다.

 

만약 이대로 세션을 사용한다면 하나의 트랜잭션 안에서 서로 다른 세션을 사용하는 문제가 발생할 수 있다.

만약 하나의 트랜잭션 안에 A, B 두 작업이 있을 때, A작업에서 사용한 세션이 정상 종료되고 B작업에 이상이 생겨 rollback 해야 할 때 A와 B는 다른 세션을 사용하므로 B만 rollback 되고 A는 commit된다. (Atomicity 보장 X)

 

이 문제는 하단 3-5. 에서 다룰 것이다.

 

 

3-3. 테이블 정의

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, BigInteger, String, DateTime, Boolean

Base = declarative_base()

class TestRecordModel(Base):
    __tablename__ = 'test_record'  # 테이블명

    cid = Column("cid", BigInteger, primary_key=True, autoincrement=True)
    id = Column("id", String(36), unique=True, nullable=False)
    keyword = Column("keyword", String(50), nullable=False)
    ...

• declarative_base()
  • SQLAlchemy는 database 테이블과 클래스의 매핑을 위해서 base 클래스를 생성해야 한다.
  • 해당 클래스를 상속받은 자식 클래스들은 모두 DB 테이블과 매핑되는 클래스로 인식된다.
• __tablename__
  • DB에서 사용하는 테이블명을 작성한다. (필수 값)

 

 

3-4. CRUD 처리

from utils import get_db_session
from sqlalchemy import select, text

from domain.test_record.model.TestRecordModel import TestRecordModel

class TestRecordRepository():

    def save(self, entity):
        get_db_session().add(entity)	# insert

    def search_list_by_workspace_id(self, workspace_id):
        query = select(TestRecordModel)\
            .where(TestRecordModel.workspace_id == workspace_id)

        return get_db_session().execute(query).all()

    def search_one(self, id):
        query = select(TestRecordModel)\
            .where(TestRecordModel.id == id)

        return get_db_session().execute(query).one()

    def soft_delete_one(self, id):
        query = text("""
            UPDATE test_record record
            SET record.deleted_flag = True
            WHERE record.id = :id
        """)
        params = {"id" : id}

        get_db_session().execute(query, params)

• `select().where()` 형식으로 쿼리를 작성한다.
• text : string 형태의 SQL을 작성할 때 사용. (SQLAlchemy Core에서 동작)
• excute() : 실행할 쿼리를 excute()인자에 넣어 실행

 

• 단일조회
  • first() : 없다면 None 리턴. 여러 개라면 한 개만 출력 (SQL의 ‘limit 1’과 같은 기능)
  • scalar() : 없다면 None 리턴. 여러 개라면 `MultipleResultsFound` 예외
  • one() : 없다면 `NoResultFound` 예외. 여러 개라면 `MultipleResultsFound` 예외
• 다중조회
  • `all()` : 쿼리의 모든 결과를 한번에 조회해 리스트로 반환한다. 각 행은 튜플로 반환된다.
  • `fetchall()`: Result 객체에서 모든 행을 가져와 리스트로 반환한다. 각 행은 튜플로 반환된다.
  • `scalars()` : ScalarResult 객체를 반환한다. ScalarResult 객체는 단일 column값을 조회하고, `unique()`, `one()`, `all()` 등의 메서드를 호출해야 전체 값을 조회할 수 있다. 따라서 scalars()만 호출하면 여러 column을 반환하는 쿼리에서 첫번째 column만 조회된다.

 

 

 

SQLAlchemy는 쿼리 작성 및 조회 방법이 다양한데, 나는 SQLAlchemy 2.0 형식으로 작성했다.

Migrating to SQLAlchemy 2.0 — SQLAlchemy 1.4 Documentation

 

쿼리 인터페이스 참고

1. `session.query()`
   • sqlalchemy version1에서 사용하는 방법
   • find(), all() 등 객체 지향적인 방식으로 데이터베이스 작업을 수행
2. `Model.query()`
   • session.query()와 동일한 방법
   • ORM 기능을 활용. 특정 모델에 대한 특정 쿼리를 작성하는 방법
3. `session.execute(select().where()~)`
   • 위 두 방법은 레거시한 작성법으로 취급되고 version2에서는 `session.execute()` 사용을 권장하고 있다.
   • 실행 결과로 ResultProxy 객체가 반환되며, 쿼리 실행 결과를 캡슐화한 객체이다. ResultProxy 객체는 one(), fetchone(), fetchall(), scalar(), all() 등과 같은 메서드로 결과를 가져올 수 있다.

 

 

3-5. 세션 관리 - commit, rollback, close

앞서 SQLAlchemy 세션은 한 요청에 대해 스레드마다 다른 세션이 생성될 수 있다고 했다. 따라서 SQLAlchemy 공식문서를 보면, 트랜잭션을 설계할 때 with문을 사용해 만들고 있다. 그럼 하나의 세션으로 여러 쿼리를 묶어 실행할 수 있다.

하지만 이 방식은 with문 안에서 트랜잭션 안에서 실행할 모든 쿼리 작업을 작성해줘야 하고, 로직이 복잡해 보일 수 있다.

따라서 나는 파이썬의 *decorator 이용해 구현했다.

 

 

* Decorator ?

데코레이터는 Python에서 제공하는 기능으로, 특정 함수의 앞뒤에서 실행될 동작을 정의할 때 사용하는 기능이다. 데코레이터를 이용한다면 공통된 로직을 간편하게 호출할 수 있고 하나의 함수는 핵심 기능에만 집중할 수 있다. @ 뒤에 실행시킬 데코레이터를 작성한다.

 

예를들어, 특정 함수의 실행 시간을 알고 싶을 때 다음과 같은 데코레이터를 사용할 수 있다.

import time

def timer_decorator(func):
    def wrapper(self, *args, **kwargs):
        start = time.time()
        result = func(self, *args, **kwargs)	# func은 test메서드를 의미
        end = time.time()

        print(f"{func.__name__} 실행시간 : {end - start:.5f}초")
        return result
    return wrapper

@timer_decorator
def test():
    sum = 0
    for i in range(1000):
        sum += i
    return sum

1. `test()` 메서드가 호출되면 decorator인 `timer_decorator()`가 시작된다.
2. start 시간을 설정하고, 실제 실행 함수인 func인 `test()`가 시작된다.
3. `test()` 메서드가 종료되면 decorator의 end 시간을 설정하고 실행시간을 계산해 출력한다.

 

 

이런 기능을 이용해 transaction을 설계했다.

def transactional(func):
    def wrapper(self, *args, **kwargs):
       try:
            results = func(self, *args, **kwargs)	# 로직 함수
            get_db_session().commit()
            return results
       except:
            get_db_session().rollback()
            raise
       finally:
            get_db_session().close()
            
    return wrapper

• `execute()`로 실행한 쿼리를 DB에 반영하려면 결국 commit() 동작이 필요하다.
• 트랜잭션으로 묶을 함수 func()은 transactional() 데코레이터 안에서 동작한다.
• func()의 실행이 마친 후, commit / rollbak 처리를 해준다.
• 항상 다 사용한 세션은 `close()` 해줘야 한다. 그렇지 않는다면 Connection Pool에 커넥션이 반환되지 않아 `TimeoutError`가 자주 발생할 수 있다.

 

 

SQLAlchemy에서 트랜잭션을 설계하는 방법에 대한 글은 다음 게시물에서 자세히 정리했다.

SQLAlchemy 트랜잭션 설계 (+ decorator)

 

 

 

4. Flask SQLAlchemy

---

Flask 웹 프레임워크와 통합하여 사용하기 위해 특별히 설계된 SQLAlchemy의 확장 라이브러리

Flask에서 Flask 구성 파일을 통해 쉽게 설정할 수 있고, 코드가 간결하다. SQLAlchemy 라이브러리를 확장해서 만들어진 것이기 때문에 사용법은 SQLAlchemy 작성 방법과 유사하다.

 

(^SQLAlchemy는 파이썬에서 데이터베이스 작업을 수행하기 위한 ORM 라이브러리이고, Flask-SQLAlchemy는 Flask 웹 프레임워크와 SQLAlchemy를 통합하기 위한 확장 라이브러리이다^)

 

 

flask-sqlalchemy 라이브러리를 사용한다면 기존의 모델 클래스를 하단과 같이 작성할 수 있다.

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, BigInteger, String

Base = declarative_base()

class TestRecordModel(Base):
    __tablename__ = 'test_record'  # 테이블명

    cid = Column("cid", BigInteger, primary_key=True, autoincrement=True)
    id = Column("id", String(36), unique=True, nullable=False)
    keyword = Column("keyword", String(50), nullable=False)
    ...

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class NRankRecordModel(db.Model):
    __tablename__ = 'test_record'  # 테이블명

    cid = db.Column("cid", db.BigInteger, primary_key=True, autoincrement=True)
    id = db.Column("id", db.String(36), unique=True, nullable=False)
    keyword = db.Column("keyword", db.String(50), nullable=False)

각각의 함수들을 직접 import 시키지 않고, db.* 처럼 바로 사용할 수 있다.

하지만 db.Column이 아닌 db.Golumn 같은 오타를 내었을 때, 에러를 표시해주지 않는다.

 

 

 

5. 마무리

---

매번 Java/Spring으로 개발을 해오다가 지난 프로젝트에서 Python/Flask을 이용했었다. 개발 프레임워크로 Flask 택했을 때, 개발하면서 생각치 못한 부분에서 많은 시행착오가 있었다. 특히 SQLAlchemy을 사용하면서 트랜잭션 처리, 세션 생성, 풀 설정 등 어느 하나 의도한 대로 돌아가지 않았다. 기술 문서나 자료들을 정리해 보고, 로그도 많이 찍어보면서 문제를 해결해 갔다. 항상 DB 관련된 처리는 어렵지만, Spring을 계속해서 사용했다면 몰랐을 기술들을 익히게 되어 유익한 시간이었다. 게다가 새로운 프레임워크를 택할 때 고려사항에 대해서도 더 폭넓게 고민해야겠다는 생각도 들었다.

 

 

 

 

참고 😃

https://ulfrid.github.io/python/python-sqlalchemy/

https://velog.io/@baeyuna97/SQLAlchemy-사용하기

https://jammdev.tistory.com/186

https://spoqa.github.io/2018/01/17/connection-pool-of-sqlalchemy.html

https://miintto.github.io/docs/python-sqlalchemy-session

https://yujuwon.tistory.com/entry/SQLALCHEMY-session-관리