datax-admin/app/models/database.py

 # database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from configs.logging import logger
from configs.settings import config

USER = config.get('DATABASE', 'USER')
PWD = config.get('DATABASE', 'pwd')
DB_NAME = config.get('DATABASE', 'DB_NAME')
HOST =  config.get('DATABASE', 'HOST')
PORT = config.get('DATABASE', 'PORT')


SQLALCHEMY_DATABASE_URL = f'mysql+mysqlconnector://{USER}:{PWD}@{HOST}:{PORT}/{DB_NAME}?charset=utf8&auth_plugin=mysql_native_password'
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, pool_pre_ping=True
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

logger.info("connect to mysql success: ")


Base = declarative_base()

class BaseModel(Base):
    __abstract__ = True
    def to_dict(self):
        return {c.name: getattr(self, c.name) for c in self.__table__.columns}