pip install SQLAlchemy
pip install aiomysql
pip install pydantic-settings
pip install python-dotenv

from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    app_name: str = "chf-app"
    database_url: str = 'mysql+aiomysql://root:chenhaifeng@47.120.55.99:3306/fastapi'
    secret: str = 'SECRET'

    class Config:
        env_file = ".env"


@lru_cache
def get_settings():
    return Settings()

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.ext.asyncio import AsyncSession
from typing import AsyncGenerator
from config.config import get_settings
import databases
from .model import Base

DATABASE_URL = get_settings().database_url
database = databases.Database(DATABASE_URL, min_size=2, max_size=5)
engine = create_async_engine(DATABASE_URL, echo=True)  # echo开启sql打印
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)


class DALGetter:
    def __init__(self, dal_cls):
        self.dal_cls = dal_cls

    async def __call__(self):
        async with async_session_maker() as session:
            async with session.begin():
                yield self.dal_cls(session)


# 获取数据库连接
async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        yield session


# 创建表
async def create_db_and_tables():
    from application.ucenter.model import User, UserRole, Role, RolePower, Power
    async with engine.begin() as conn:
        if get_settings().drop_all_table:
            await conn.run_sync(Base.metadata.drop_all)
        if get_settings().create_all_table:
            await conn.run_sync(Base.metadata.create_all)

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.ext.asyncio import AsyncAttrs

class Base(AsyncAttrs, DeclarativeBase):
    pass

from fastapi import APIRouter

from sqlalchemy import select
from sqlalchemy.orm import selectinload
from extensions.db.mysql import get_async_session
from application.ucenter.model import Role, Power
from .schema import GetRoleDetail, AddRole, UpdateRole, AddPowerReq, AddPowerResp, RemodePowerReq, RemodePowerResp
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import Depends

role_router = APIRouter()


@role_router.get("/detail", response_model=GetRoleDetail)
async def get_role(pk: int, session: AsyncSession = Depends(get_async_session)):
    """
    获取单个角色
    """
    query = select(Role).where(Role.id == pk)
    result = await session.execute(query)
    return result.unique().scalar_one_or_none()


@role_router.post("/add")
async def add_role(data: AddRole, session: AsyncSession = Depends(get_async_session)):
    role = Role(**data.model_dump())
    session.add(role)
    await session.commit()
    await session.refresh(role)
    return {"msg": "ok", "id": role.id}


@role_router.put("/update")
async def update_role(pk: int, data: UpdateRole, session: AsyncSession = Depends(get_async_session)):
    query = select(Role).where(Role.id == pk)
    res = await session.execute(query)
    role = res.unique().scalar_one_or_none()
    for k, v in data.model_dump().items():
        setattr(role, k, v)
    session.add(role)
    await session.commit()
    await session.refresh(role)
    return {"msg": "ok", "id": role.id}


@role_router.delete("/delete")
async def delete_role(pk: int, session: AsyncSession = Depends(get_async_session)):
    query = select(Role).where(Role.id == pk)
    res = await session.execute(query)
    role = res.unique().scalar_one_or_none()
    session.delete(role)
    await session.commit()
    return {"msg": "ok", "id": pk}


@role_router.post("/add/powers", response_model=AddPowerResp)
async def add_powers(req: AddPowerReq, session: AsyncSession = Depends(get_async_session)):
    """
        给角色(role) 添加权限(power)
    """
    # 查询power是否存在
    power_query = select(Power).where(Power.id.in_(req.power_list))
    power_resp = await session.execute(power_query)

    # 查询role是否存在,且预加载多对多powers字段(添加时才能触发,自动查重复添加)
    role_resp = await session.scalars(select(Role).where(Role.id == req.role_id).options(selectinload(Role.powers)))
    role = role_resp.first()
    role.powers = power_resp.scalars().all()
    session.add(role)
    await session.commit()

    return AddPowerResp(id=1, msg="ok")


@role_router.post("/remode/powers", response_model=RemodePowerResp)
async def remode_powers(req: RemodePowerReq, session: AsyncSession = Depends(get_async_session)):
    """
        角色(role) 移除权限(power)
    """
    power_query = select(Power).where(Power.id.in_(req.power_list))
    power_resp = await session.execute(power_query)

    role_resp = await session.scalars(select(Role).where(Role.id == req.role_id).options(selectinload(Role.powers)))
    role = role_resp.first()
    for power in power_resp.scalars():
        role.powers.remove(power)
    await session.commit()
    return RemodePowerResp(msg="ok")


@role_router.post("/test")
async def test_role(session: AsyncSession = Depends(get_async_session)):
    async with session.begin():
        session.add_all(
            [
                Role(name="角色1"),
                Role(name="角色2"),
                Role(name="角色3"),
            ]
        )
    await session.commit()
    return {"msg": "ok"}