[Python] SQLAlchemy ORM(MySql) 사용하는 방법

Java에서 JPA를 이용해서 사용하다보니 Python을 사용할 때도 ORM을 사용합니다.

예제들이 Flask를 기반으로 ORM사용을 하는게 많았는데 SQLAlchemy만을 이용합니다.

SQLAlchemy 섪치

pip3 install sqlalchemy

pip3 install mysqlclient

sqlalchemy와 mysqlclient를 pip를 이용해 설치합니다.

드라이버(Engin) 및 세션(Session) 로드

from sqlalchemy import create_engine
engine = create_engine('mysql+mysqldb://<username>:<password>@<host>:<port>/<dbname>')

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(engine) # Engine에 종속적인 Session 정의
session = Session() # Session 객체 생성

DB를 연결합니다.
세션은 DB와 소통을 관리하는 영역을 의미하므로 연결하여 사용하도록 합니다.
sessionmaker는 session을 만들어주는 팩토리 클래스입니다.

테이블(table) 클래스 정의

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()


import datetime
from sqlalchemy import Column, Integer, String, DateTime
class User(Base):
    __tablename__ = '<tablename>'
    a = Column(Integer)
    b = Column(Integer)
    time = Column(DateTime, default=datetime.datetime.utcnow, primary_key=True)
    def __init__(self, a, b):
        self.a = a
        self.b = b

Base 클래스에서 DDL을 기본적으로 제공하고 생성이나 삭제나 업데이트를 수행합니다.
이를 상속받아 사용하면 DB와 연결된 클래스를 만들 수 있습니다.

Base.metadata.create_all(engine) # Base에 연결된 모든 테이블을 DB에 생성
Base.metadata.tables['TEST_TB'].create(bind = db._engine) # Base에 연결된 TEST_TB 테이블을 DB에 생성
Base.metadata.drop_all(engine) # Base에 연결된 모든 테이블을 DB에서 제거
Base.metadata.tables['TEST_TB'].drop(bind = db._engine) # Base에 연결된 TEST_TB 테이블을 DB에서 제거

쿼리(Query) 및 커밋(Commit)

insert

user = User(1,2)
session.add(user)
session.commit()

샘플 코드

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASES = create_engine('mysql+mysqldb://<user>:<password>@<localhost>:<port>:<databaseName>', echo = True)

# orm과의 매핑 선언
Base = declarative_base()

class Test(Base):

   __tablename__ = 'test'

   id       = Column(Integer, primary_key=True)
   name     = Column(String(50))
   fullname = Column(String(50))
   password = Column(String(50))

   def __init__(self, name, fullname, password):

       self.name     = name
       self.fullname = fullname
       self.password = password

   def __repr__(self):
       return "<Test('%s', '%s', '%s')>" % (self.name, self.fullname, self.password)

if __name__ == '__main__':

   Base.metadata.create_all(DATABASES)

   # 세션 연결
   Session = sessionmaker()
   Session.configure(bind=DATABASES)
   session = Session()


   test = Test('t','t','t')

   # 세션에 추가를 한다.
   session.add(test)
   session.commit()