group_xinghuo_jinrong/app/config/database.py

"""MySQL / Redis 连接工厂。"""

from functools import lru_cache

from sqlalchemy import create_engine
from sqlalchemy.engine import Engine

from app.config.settings import settings


def _mysql_url(database: str) -> str:
    pwd = settings.mysql_password
    auth = f"{settings.mysql_user}:{pwd}" if pwd else settings.mysql_user
    return (
        f"mysql+pymysql://{auth}@{settings.mysql_host}:{settings.mysql_port}"
        f"/{database}?charset=utf8mb4"
    )


@lru_cache
def get_agent_engine() -> Engine:
    return create_engine(_mysql_url(settings.mysql_database), pool_pre_ping=True)


@lru_cache
def get_core_engine() -> Engine:
    return create_engine(_mysql_url(settings.mysql_core_database), pool_pre_ping=True)