mirror of
https://github.com/PaiGramTeam/PaiGram.git
synced 2024-11-25 18:04:10 +00:00
8f424bf0d4
♻️ 重构插件系统 ⚙️ 重写插件 🎨 改进代码结构 📝 完善文档 Co-authored-by: zhxy-CN <admin@owo.cab> Co-authored-by: 洛水居室 <luoshuijs@outlook.com> Co-authored-by: xtaodada <xtao@xtaolink.cn> Co-authored-by: Li Chuangbo <im@chuangbo.li>
31 lines
1.0 KiB
Python
31 lines
1.0 KiB
Python
from sqlalchemy.ext.asyncio import create_async_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
from typing_extensions import Self
|
|
|
|
from core.config import BotConfig
|
|
from core.service import Service
|
|
|
|
|
|
class MySQL(Service):
|
|
@classmethod
|
|
def from_config(cls, config: BotConfig) -> Self:
|
|
return cls(**config.mysql.dict())
|
|
|
|
def __init__(self, host: str = "127.0.0.1", port: int = 3306, username: str = "root", # nosec B107
|
|
password: str = "", database: str = ""): # nosec B107
|
|
self.database = database
|
|
self.password = password
|
|
self.user = username
|
|
self.port = port
|
|
self.host = host
|
|
self.engine = create_async_engine(
|
|
f"mysql+asyncmy://{self.user}:{self.password}@{self.host}:{self.port}/{self.database}"
|
|
)
|
|
self.Session = sessionmaker(bind=self.engine, class_=AsyncSession)
|
|
|
|
async def get_session(self):
|
|
"""获取会话"""
|
|
async with self.Session() as session:
|
|
yield session
|