[SAMPLE] SQLAlchemy Setup

sqlalchemy_connection.py

from sqlalchemy.engine.url import URL
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

CORE = int(os.cpu_count())

DB_URL = URL.create(
    drivername="mysql+pymysql",
    username='root',
    password="1234",
    database="",
    host="127.0.0.1",
    port=1234
)

db_engine = engine.create_engine(
    DB_URL,
    pool_pre_ping=True,
    pool_recycle=3600,
    pool_size=5,
    max_overflow=5,
    pool_timeout=10,
    echo=True
)

db_session: Session = scoped_session(sessionmaker(
    bind=db_engine,
    expire_on_commit=False,
    autocommit=False,
    autoflush=False
))
    

def transactional(func: Callable):  # XXX: transactionl 임시구현, 이 방법(decorator)은 type hint가 적용이 안됨
    @functools.wraps(func)
    def _wrapper(*args, **kwargs):
        try:
            result = func(*args, **kwargs)
            db_session.flush()
            db_session.commit()
            return result
        except Exception as e:
            db_session.rollback() 
            raise e
        finally:
            db_session.close()
    return _wrapper