opsxsql/backend/app/api/v1/endpoints/redis.py

"""Redis management API endpoints."""
from typing import Optional

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_db
from app.core.security import get_current_user, require_role
from app.models.user import User
from app.models.instance import Instance
from app.services import redis_connector

router = APIRouter(tags=["Redis管理"])


async def _get_redis_instance(instance_id: int, db: AsyncSession) -> Instance:
    """Fetch and validate a Redis instance."""
    result = await db.execute(select(Instance).where(Instance.id == instance_id))
    instance = result.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="数据源不存在")
    if instance.db_type != "redis":
        raise HTTPException(status_code=400, detail="该实例不是 Redis 类型")
    return instance


# ── Request schemas ──────────────────────────────────────────────

class RedisKeyCreate(BaseModel):
    db: int = 0
    key: str
    value: object
    type: str = "string"
    ttl: Optional[int] = None


class RedisCommandRequest(BaseModel):
    db: int = 0
    command: str


class RedisExpireRequest(BaseModel):
    db: int = 0
    seconds: int


# ── Endpoints ────────────────────────────────────────────────────

@router.get("/redis/{instance_id}/info")
async def get_redis_info(
    instance_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Redis 信息概览."""
    instance = await _get_redis_instance(instance_id, db)
    try:
        info = await redis_connector.RedisConnector.get_info(instance)
        return info
    except Exception as e:
        raise HTTPException(status_code=500, detail="获取 Redis 信息失败: " + str(e))


@router.get("/redis/{instance_id}/dbsize")
async def get_redis_dbsize(
    instance_id: int,
    db_num: int = Query(0, alias="db"),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """DBSIZE."""
    instance = await _get_redis_instance(instance_id, db)
    try:
        size = await redis_connector.RedisConnector.get_dbsize(instance, db_num)
        return {"db": db_num, "size": size}
    except Exception as e:
        raise HTTPException(status_code=500, detail="获取 DBSIZE 失败: " + str(e))


@router.get("/redis/{instance_id}/keys")
async def get_redis_keys(
    instance_id: int,
    db_num: int = Query(0, alias="db"),
    pattern: str = Query("*"),
    type: Optional[str] = Query(None, alias="type"),
    cursor: int = Query(0),
    count: int = Query(50),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """键列表."""
    instance = await _get_redis_instance(instance_id, db)
    try:
        next_cursor, keys = await redis_connector.RedisConnector.scan_keys(
            instance, db=db_num, pattern=pattern,
            type_filter=type, cursor=cursor, count=count,
        )
        return {"cursor": next_cursor, "keys": keys}
    except Exception as e:
        raise HTTPException(status_code=500, detail="获取键列表失败: " + str(e))


@router.get("/redis/{instance_id}/keys/{key:path}")
async def get_redis_key_detail(
    instance_id: int,
    key: str,
    db_num: int = Query(0, alias="db"),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """键详情."""
    instance = await _get_redis_instance(instance_id, db)
    try:
        detail = await redis_connector.RedisConnector.get_key_value(
            instance, db_num, key
        )
        return detail
    except Exception as e:
        raise HTTPException(status_code=500, detail="获取键详情失败: " + str(e))


@router.post("/redis/{instance_id}/keys")
async def create_redis_key(
    instance_id: int,
    data: RedisKeyCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(require_role("admin", "dba")),
):
    """创建/更新键."""
    instance = await _get_redis_instance(instance_id, db)
    try:
        result = await redis_connector.RedisConnector.set_key_value(
            instance, data.db, data.key, data.value, data.type, data.ttl
        )
        return {"message": "操作成功", "result": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail="创建键失败: " + str(e))


@router.delete("/redis/{instance_id}/keys/{key:path}")
async def delete_redis_key(
    instance_id: int,
    key: str,
    db_num: int = Query(0, alias="db"),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(require_role("admin", "dba")),
):
    """删除键."""
    instance = await _get_redis_instance(instance_id, db)
    try:
        deleted = await redis_connector.RedisConnector.del_key(instance, db_num, key)
        return {"message": "删除成功", "deleted": deleted}
    except Exception as e:
        raise HTTPException(status_code=500, detail="删除键失败: " + str(e))


@router.put("/redis/{instance_id}/keys/{key:path}/expire")
async def set_redis_key_expire(
    instance_id: int,
    key: str,
    data: RedisExpireRequest,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(require_role("admin", "dba")),
):
    """设置 TTL."""
    instance = await _get_redis_instance(instance_id, db)
    try:
        ok = await redis_connector.RedisConnector.set_ttl(
            instance, data.db, key, data.seconds
        )
        return {"message": "设置成功" if ok else "设置失败"}
    except Exception as e:
        raise HTTPException(status_code=500, detail="设置 TTL 失败: " + str(e))


@router.delete("/redis/{instance_id}/flushdb")
async def flush_redis_db(
    instance_id: int,
    db_num: int = Query(0, alias="db"),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(require_role("admin")),
):
    """清空DB (admin only!)."""
    instance = await _get_redis_instance(instance_id, db)
    try:
        result = await redis_connector.RedisConnector.flush_db(instance, db_num)
        return {"message": "清空成功", "result": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail="清空 DB 失败: " + str(e))


@router.post("/redis/{instance_id}/command")
async def execute_redis_command(
    instance_id: int,
    data: RedisCommandRequest,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(require_role("admin", "dba")),
):
    """执行命令."""
    instance = await _get_redis_instance(instance_id, db)
    try:
        output = await redis_connector.RedisConnector.execute_command(
            instance, data.db, data.command
        )
        return {"output": output}
    except Exception as e:
        raise HTTPException(status_code=500, detail="命令执行失败: " + str(e))


@router.get("/redis/{instance_id}/slowlog")
async def get_redis_slowlog(
    instance_id: int,
    count: int = Query(10),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """慢日志."""
    instance = await _get_redis_instance(instance_id, db)
    try:
        logs = await redis_connector.RedisConnector.get_slowlog(instance, count)
        return logs
    except Exception as e:
        raise HTTPException(status_code=500, detail="获取慢日志失败: " + str(e))


@router.get("/redis/{instance_id}/clients")
async def get_redis_clients(
    instance_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """客户端列表."""
    instance = await _get_redis_instance(instance_id, db)
    try:
        clients = await redis_connector.RedisConnector.get_clients(instance)
        return clients
    except Exception as e:
        raise HTTPException(status_code=500, detail="获取客户端列表失败: " + str(e))