Agent/examples/piaoquan_demand/pattern_toos/db_manager.py

from sqlalchemy import create_engine, and_, or_, desc
from sqlalchemy.orm import sessionmaker, Session


class DatabaseManager1:
    """数据库管理类"""

    # mysql+pymysql://<用户名>:<密码>@<主机地址>:<端口>/<数据库名>?charset=utf8mb4
    def __init__(self):
        connection_string = (
            f"mysql+pymysql://wx2016_longvideo:wx2016_longvideoP%40assword1234@rm-bp1k5853td1r25g3n690.mysql.rds.aliyuncs.com:3306/open_aigc?charset=utf8mb4"
            # f"mysql+pymysql://root:aigc_admin@127.0.0.1:3306/open_aigc_pattern?charset=utf8mb4"
        )
        self.engine = create_engine(connection_string, pool_pre_ping=True, pool_recycle=3600)
        self.SessionLocal = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)

    def get_session(self) -> Session:
        """获取数据库会话"""
        return self.SessionLocal()


class DatabaseManager2:
    """数据库管理类"""

    # mysql+pymysql://<用户名>:<密码>@<主机地址>:<端口>/<数据库名>?charset=utf8mb4
    def __init__(self):
        connection_string = (
            f"mysql+pymysql://root:%40Mm789654123@127.0.0.1:3306/open_aigc_pattern?charset=utf8mb4"
            # f"mysql+pymysql://root:aigc_admin@127.0.0.1:3306/open_aigc_pattern?charset=utf8mb4"
        )
        self.engine = create_engine(connection_string, pool_pre_ping=True, pool_recycle=3600)
        self.SessionLocal = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)

    def get_session(self) -> Session:
        """获取数据库会话"""
        return self.SessionLocal()