app/core/scripts.py

from __future__ import annotations

import logging
from pathlib import Path
from typing import Dict, Optional

from redis.asyncio import Redis

logger = logging.getLogger(__name__)

LUA_DIR = Path(__file__).resolve().parent.parent.parent / "lua"


def load_lua(filename: str) -> str:
    path = LUA_DIR / filename
    return path.read_text(encoding="utf-8")


async def load_scripts(redis: Optional[Redis]) -> Dict[str, str]:
    if not redis:
        logger.warning("Redis not available, Lua scripts not loaded")
        return {}
    acquire_slot_sha = await redis.script_load(load_lua("acquire_slot.lua"))
    lock_key_sha = await redis.script_load(load_lua("lock_key.lua"))
    logger.info("Lua scripts loaded")
    return {
        "acquire_slot_sha": acquire_slot_sha,
        "lock_key_sha": lock_key_sha,
    }