smolkik-code
7c7c88621d
- FastAPI REST API with JWT auth - aiogram 3 Telegram bot with admin middleware - APScheduler daily tasks (expiry, reminders, revoke, sync) - SQLAlchemy 2 async ORM with Alembic migrations - Jinja2 admin panel (Dashboard, Users, Payments, Servers, Tariffs) - VPN provider abstraction with MockProvider - Stats service with revenue/subscription analytics - Docker Compose (PostgreSQL + Redis + app) - Healthcheck endpoint
37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
from typing import Any, Awaitable, Callable
|
|
|
|
from aiogram import BaseMiddleware
|
|
from aiogram.types import Message
|
|
|
|
from app.database import async_session_factory
|
|
from app.repositories.admin import AdminRepository
|
|
from app.bot.texts import ADMIN_ONLY
|
|
|
|
|
|
class AdminMiddleware(BaseMiddleware):
|
|
async def __call__(
|
|
self,
|
|
handler: Callable[[Message, dict[str, Any]], Awaitable[Any]],
|
|
event: Message,
|
|
data: dict[str, Any],
|
|
) -> Any:
|
|
if not isinstance(event, Message):
|
|
return await handler(event, data)
|
|
|
|
async with async_session_factory() as session:
|
|
repo = AdminRepository(session)
|
|
admin = await repo.get_by_telegram_id(event.from_user.id)
|
|
|
|
if admin is None or not admin.is_active:
|
|
await event.answer(ADMIN_ONLY, parse_mode="HTML")
|
|
return
|
|
|
|
data["admin"] = {
|
|
"admin_id": admin.id,
|
|
"telegram_id": admin.telegram_id,
|
|
"role": admin.role.value,
|
|
"username": admin.username,
|
|
}
|
|
|
|
return await handler(event, data)
|