[SQLAlchemy] declarative_base 로부터 DB에 테이블 생성하기

목차

 

개요 

django의 manage.py에서 migrate에 관련된 기능을 보면 django 내부에 선언된 Model로부터 DataBase에 Migrate를 해주는 명령이 존재한다.

$ python manage.py makemigrations
$ python manage.py migrate

django에서는 위와 같은 방식으로 정의한 ORM을 DB에다 테이블을 생성 혹은 업데이트 해준다. SQLAlChemy에서도 이와 비슷한 방식을 지원 도구가 있는데 alembic이다.

 

그런데 alembic을 사용하지 않고 declarative_base로 선언된 model로 SQL Statement를 추출하고 이를 어떻게 DB 반영할 수 있을까?

 

 

alembic ? 굳이 ?

sqlalchmey에서 declarative_base를 사용하게되면 다음과 같은 형식의 코드가 나올 것이다. 다음은 declarative_base를 다루는 기본적인 예제이다.

from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Table01(Base): ...
class Table02(Base): ...
class Table03(Base): ...

위 코드는 특정 도구를 사용하지 않는 이상 위 코드를 수정한다고 해서 DataBase에 변경을 가하는 행위는 일어나지 않는다는 점이다.

앞서 언급했듯 DataBase 변경관리를 해주는 Alembic이라는 도구가 상황에 가장 적합해 보인다. 하지만 나는 local에서 Test용으로 DB를 생성하고 최초 한 번만 Test용 DB에다 테이블을 생성해주면 되는 상황이기에 굳이 Alembic을 사용할 필요가 없었다.

 

---

예제로 파악해보자.

이제 declarative_base로부터 Table을 생성하는 SQL Statement를 뽑아보자. SQLALchemy에서 제공하는 도구를 이용하자.

from sqlalchemy.schema import CreateTable

예를 들어 다음과 같이 declararive_base로 선언된 테이블 정보가 있다고 가정해 보자.

from sqlalchemy import Column, String, CHAR

from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Table(Base):
    __tablename__ = 'sample_table'

    uuid = Column('uuid', CHAR(36), primary_key=True)
    name = Column('name', CHAR(12), nullable=False)

위 테이블에서 sqlalchemy.schema.CreateTable을 이용하여 Create SQL Statement를 얻어로면 다음과 같은 순서로 접근해야 한다.

for table in Base.metadata.tables.values():
        
    _statements = CreateTable(table).compile(dialect=mysql.dialect())
	print(_statements)
    

# Result
CREATE TABLE sample_table (
	uuid CHAR(36) NOT NULL, 
	name CHAR(12) NOT NULL, 
	PRIMARY KEY (uuid)
)

주의할 점은 compile 메서드에서 자신이 사용하는 DB에 맞는 dialect를 지정하면 된다. 전체 코드를 보면 다음과 같이 생겨먹었다.

from __future__ import annotations

from sqlalchemy import Column, CHAR
from sqlalchemy.dialects import mysql
from sqlalchemy.orm import declarative_base
from sqlalchemy.schema import CreateTable

Base = declarative_base()


class Table(Base):
    __tablename__ = 'sample_table'

    uuid = Column('uuid', CHAR(36), primary_key=True)
    name = Column('name', CHAR(12), nullable=False)


if __name__ == '__main__':

    for table in Base.metadata.tables.values():
        print(table, type(table))

        _statements = CreateTable(table).compile(dialect=mysql.dialect())
        print(_statements)

 

---

IDE에서 sql을 읽는 데 실패함.

필자는 DBeaver라는 DB IDE를 사용 중인데 위에서 SQL Statement를 생성한 것으로 DBeaver에 복사 붙여 넣기 해서 sql을 실행하면 실패한다. sql을 해석하는 과정에서 오류를 뱉어내는 것으로 확인했는데 Table의 특정 속성을 인식하지 못하는 문제였다 속성을 백 틱(`)으로 감싸주거나 하나씩 실행해주면 문제가 해결되는데 Schema의 수가 많을수록 적용하는 시간도 오래 걸린다.  

이 문제를 해결하기 위해 CreateTable이라는 Library에서 백틱(`) 처리를 해주는 옵션이 존재하는지 찾아보긴 했는데 찾을 수 없었다. 그러니 다른 방법을 SQLAlchemy의 session을 바탕으로 execute() 함수를 실행해주면 문제없이 해결된다.

테이블의 개수가 늘어난 경우 다음과 같이 list에 table의 정보를 str로 변환한 뒤 넣어준다.

Base = declarative_base()


class Table(Base):
    __tablename__ = 'sample_table'

    uuid = Column('uuid', CHAR(36), primary_key=True)
    name = Column('name', CHAR(12), nullable=False)


class OtherTable(Base):
    __tablename__ = 'sample_other_table'

    uuid = Column('uuid', CHAR(36), primary_key=True)
    name = Column('name', CHAR(12), nullable=False)


stmts = []

for table in Base.metadata.tables.values():
    _create_sql = str(CreateTable(table).compile(dialect=mysql.dialect()))

    stmts.append(_create_sql)

for x in stmts:
    print(x)

이후부터는 다음과 같이 SQLAlchemy에서 DB Session을 얻어 list에 들어간 Create SQL statement를 하나씩 실행해주면 된다.

# 의사코드, 실제로 동작안할 수 있음

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine(self.url)

session = sessionmaker()
session.configure(bind=engine)


for create_sql in create_table_statements:
    session().execute(create_sql)

session().commit()
session().close()