1. SQLAlchemy란?

SQLAlchemy란?

• python으로 database에 연결하는 ORM중 하나입니다. ORM(Object-Related Manager)로 객체 관계 매핑을 의미하며, 호환되지 않는 타입 시스템간에 데이터를 변환하기 위해 사용하는 프로그래밍 기술입니다.
• 애플리케이션 코드나 데이터베이스 시스템에 간섭하지 않는 다는 특징이 있습니다.

2. SQLAlchemy의 장점

모델과 데이터베이스를 한번만 정의

• 데이터베이스 파일과 모델 파일을 분리하여, 필요할 때마다 불러올 수 있습니다.
• asyncio를 지원하여 여러 작업을 동시에 작업이 가능하다.

3. Session과 engine 생성 및 연결

mysql 데이터베이스에 연결하는 session 생성 및 engine과 연결

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# SQLAlchemy 엔진 생성
engine = create_engine(f"mysql+pymysql://{user}:{password}@{host}/{db}?charset={charset}")
Session = sessionmaker(bind=engine)
self.session = Session()

4. Session 닫기

mysql 데이터베이스에 연결하는 session 닫기

session.close()

5. Connection 전체 코드

Class로 만들어 필요할 때 마다 사용가능하게 불러오겠습니다. return은 session입니다.

• pydoc을 통해 해당 함수에 대한 설명과 param과 return 등을 적어 해당 함수 사용시 알아보기 편하게 생활화하고 있습니다.
• try, except와 logger를 통해 해당 부분에서 문제가 발생할 경우, 쉽게 알아볼 수 있게 작성하였습니다.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import SQLAlchemyError

class DbConnRepository:
    def __init__(self, mysql_db_connection, logger):
        """
        Database connection repository

        :param mysql_db_connection: MySQL database connection
        :param logger: logging logger
        """
        self.session = None
        self.mysql_db_connection = mysql_db_connection
        self.logger = logger

    def get_mysql_connection(self):
        """
        CREATE SQLAlchemy engine and MySQL database connection

        :return: SQLAlchemy engine session
        """
        host = self.mysql_db_connection['host']
        user = self.mysql_db_connection['user']
        password = self.mysql_db_connection['password']
        db = self.mysql_db_connection['db']
        charset = self.mysql_db_connection['charset']

        try:
            # SQLAlchemy 엔진 생성
            engine = create_engine(f"mysql+pymysql://{user}:{password}@{host}/{db}?charset={charset}")
            Session = sessionmaker(bind=engine)
            self.session = Session()
            self.logger.info(f"[CONNECT] get_mysql_connection")
            return self.session
        except SQLAlchemyError as e:
            self.logger.error(
                f"[ERROR] get_mysql_connection\n"
                f"Failed to connect to MySQL: {e}"
            )
            raise

    def close_mysql_connection(self):
        """
        CLOSE MySQL Session connection
        """
        if self.session:
            try:
                self.session.close()
                self.logger.info(f"[CLOSE] get_mysql_connection: session")
            except Exception as e:
                self.logger.error(
                    f"[ERROR] close_mysql_connection\n"
                    f"Failed to close MySQL session: {e}"
                )
                raise

6. References

1. ks1171-park blog
2. python docs-asyncio