Agent/examples/demand/db_manager.py

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


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

    # mysql+pymysql://<用户名>:<密码>@<主机地址>:<端口>/<数据库名>?charset=utf8mb4
    def __init__(self):
        connection_string = (
            f"mysql+pymysql://content_rw:bC1aH4bA1lB0@rm-t4nh1xx6o2a6vj8qu3o.mysql.singapore.rds.aliyuncs.com: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()