upd project structure

This commit is contained in:
Alexander Kalinovsky
2025-02-18 20:57:52 +01:00
parent baa55d28d6
commit 1d162677a8
62 changed files with 1305 additions and 48 deletions

22
src/qbot/__init__.py Normal file
View File

@@ -0,0 +1,22 @@
from .main import QBotApp as QBotApp
from .config import Config as Config
from .router import Router as Router
from .model.bot_entity import BotEntity as BotEntity
from .model.bot_enum import BotEnum as BotEnum, EnumMember as EnumMember
from .bot.handlers.context import (
ContextData as ContextData,
CallbackCommand as CallbackCommand,
CommandContext as CommandContext,
)
from .model.descriptors import (
Entity as Entity,
EntityField as EntityField,
FormField as FormField,
EntityForm as EntityForm,
EntityList as EntityList,
EntityPermission as EntityPermission,
CommandCallbackContext as CommandCallbackContext,
CommandButton as CommandButton,
FieldEditButton as FieldEditButton,
InlineButton as InlineButton,
)

View File

@@ -0,0 +1,39 @@
from typing import Annotated
from fastapi import APIRouter, Depends, Request, Response
from sqlmodel.ext.asyncio.session import AsyncSession
from ..main import QBotApp
from ..db import get_db
from aiogram.types import Update
from logging import getLogger
logger = getLogger(__name__)
logger.setLevel("DEBUG")
router = APIRouter()
@router.post("/webhook")
async def telegram_webhook(
db_session: Annotated[AsyncSession, Depends(get_db)], request: Request
):
logger.debug("Webhook request %s", await request.json())
app: QBotApp = request.app
request_token = request.headers.get("X-Telegram-Bot-Api-Secret-Token")
if request_token != app.bot_auth_token:
logger.warning("Unauthorized request %s", request)
return Response(status_code=403)
try:
update = Update(**await request.json())
except Exception:
logger.error("Invalid request", exc_info=True)
return Response(status_code=400)
try:
await app.dp.feed_webhook_update(
app.bot, update, db_session=db_session, app=app
)
except Exception:
logger.error("Error processing update", exc_info=True)
return Response(status_code=200)

14
src/qbot/auth/__init__.py Normal file
View File

@@ -0,0 +1,14 @@
from ..model.settings import Settings
from ..model.user import UserBase
from ..bot.handlers.context import ContextData, CallbackCommand, CommandContext
async def authorize_command(user: UserBase, callback_data: ContextData):
if (
callback_data.command == CallbackCommand.MENU_ENTRY_PARAMETERS
or callback_data.context == CommandContext.SETTING_EDIT
):
allowed_roles = await Settings.get(Settings.SECURITY_PARAMETERS_ROLES)
return any(role in user.roles for role in allowed_roles)
return False

View File

@@ -0,0 +1,26 @@
from aiogram.filters import Filter
from aiogram.fsm.context import FSMContext
from .handlers.context import CallbackCommand, ContextData
from logging import getLogger
logger = getLogger(__name__)
class CallbackCommandFilter(Filter):
def __init__(self, command: CallbackCommand):
self.command = command
async def __call__(self, *args, **kwargs):
state: FSMContext = kwargs.get("state")
state_data = await state.get_data()
context_data = state_data.get("context_data")
if context_data:
try:
context_data = ContextData.unpack(context_data)
except Exception:
logger.error("Error unpacking context data", exc_info=True)
return False
else:
return context_data.command == self.command
return False

View File

View File

@@ -0,0 +1,37 @@
from aiogram.types import InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from ....model.descriptors import EntityDescriptor
from ....utils.main import get_callable_str
from ..context import ContextData, CallbackCommand
def add_filter_controls(
keyboard_builder: InlineKeyboardBuilder,
entity_descriptor: EntityDescriptor,
filter: str = None,
filtering_fields: list[str] = None,
page: int = 1,
):
caption = ", ".join(
[
get_callable_str(
entity_descriptor.fields_descriptors[field_name].caption,
entity_descriptor,
)
if entity_descriptor.fields_descriptors[field_name].caption
else field_name
for field_name in filtering_fields
]
)
keyboard_builder.row(
InlineKeyboardButton(
text=f"🔎 {caption}{f': "{filter}"' if filter else ''}",
callback_data=ContextData(
command=CallbackCommand.VIEW_FILTER_EDIT,
entity_name=entity_descriptor.name,
data=str(page),
).pack(),
)
)

View File

@@ -0,0 +1,168 @@
from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from sqlmodel.ext.asyncio.session import AsyncSession
from typing import TYPE_CHECKING
from ..context import ContextData, CallbackCommand
from ...command_context_filter import CallbackCommandFilter
from ....model.user import UserBase
from ....model.settings import Settings
from ....model.view_setting import ViewSetting
from ....utils.main import (
get_send_message,
get_entity_descriptor,
get_field_descriptor,
)
from ....utils.serialization import deserialize
from ..editors.entity import render_entity_picker
from .routing import route_callback
if TYPE_CHECKING:
from ....main import QBotApp
router = Router()
@router.callback_query(
ContextData.filter(F.command == CallbackCommand.VIEW_FILTER_EDIT)
)
async def view_filter_edit(query: CallbackQuery, **kwargs):
callback_data: ContextData = kwargs["callback_data"]
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
args = callback_data.data.split("&")
page = int(args[0])
cmd = None
if len(args) > 1:
cmd = args[1]
db_session: AsyncSession = kwargs["db_session"]
app: "QBotApp" = kwargs["app"]
user: UserBase = kwargs["user"]
entity_descriptor = get_entity_descriptor(app=app, callback_data=callback_data)
if cmd in ["cancel", "clear"]:
if cmd == "clear":
await ViewSetting.set_filter(
session=db_session,
user_id=user.id,
entity_name=entity_descriptor.class_name,
filter=None,
)
context_data_bak = state_data.pop("context_data_bak", None)
if context_data_bak:
state_data["context_data"] = context_data_bak
context_data = ContextData.unpack(context_data_bak)
field_descriptor = get_field_descriptor(app, context_data)
edit_prompt = state_data["edit_prompt"]
current_value = await deserialize(
session=db_session,
type_=field_descriptor.type_,
value=state_data["value"],
)
page = int(state_data.pop("page"))
kwargs.pop("callback_data")
return await render_entity_picker(
field_descriptor=field_descriptor,
message=query,
callback_data=context_data,
current_value=current_value,
edit_prompt=edit_prompt,
page=page,
**kwargs,
)
else:
state_data.pop("context_data", None)
return await route_callback(message=query, back=False, **kwargs)
# await save_navigation_context(callback_data = callback_data, state = state)
old_context_data = state_data.get("context_data")
await state.update_data(
{
"context_data": callback_data.pack(),
"context_data_bak": old_context_data,
"page": page,
}
)
send_message = get_send_message(query)
await send_message(
text=await Settings.get(Settings.APP_STRINGS_VIEW_FILTER_EDIT_PROMPT),
reply_markup=InlineKeyboardBuilder()
.row(
InlineKeyboardButton(
text=await Settings.get(Settings.APP_STRINGS_CANCEL_BTN),
callback_data=ContextData(
command=CallbackCommand.VIEW_FILTER_EDIT,
entity_name=entity_descriptor.name,
data=f"{page}&cancel",
).pack(),
),
InlineKeyboardButton(
text=await Settings.get(Settings.APP_STRINGS_CLEAR_BTN),
callback_data=ContextData(
command=CallbackCommand.VIEW_FILTER_EDIT,
entity_name=entity_descriptor.name,
data=f"{page}&clear",
).pack(),
),
)
.as_markup(),
)
@router.message(CallbackCommandFilter(command=CallbackCommand.VIEW_FILTER_EDIT))
async def view_filter_edit_input(message: Message, **kwargs):
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
callback_data = ContextData.unpack(state_data["context_data"])
db_session: AsyncSession = kwargs["db_session"]
user: UserBase = kwargs["user"]
app: "QBotApp" = kwargs["app"]
entity_descriptor = get_entity_descriptor(app=app, callback_data=callback_data)
filter = message.text
await ViewSetting.set_filter(
session=db_session,
user_id=user.id,
entity_name=entity_descriptor.class_name,
filter=filter,
)
# state_data.pop("context_data")
# return await route_callback(message = message, back = False, **kwargs)
context_data_bak = state_data.pop("context_data_bak", None)
if context_data_bak:
state_data["context_data"] = context_data_bak
context_data = ContextData.unpack(context_data_bak)
field_descriptor = get_field_descriptor(app, context_data)
edit_prompt = state_data["edit_prompt"]
current_value = await deserialize(
session=db_session, type_=field_descriptor.type_, value=state_data["value"]
)
page = int(state_data.pop("page"))
return await render_entity_picker(
field_descriptor=field_descriptor,
message=message,
callback_data=context_data,
current_value=current_value,
edit_prompt=edit_prompt,
page=page,
**kwargs,
)
else:
state_data.pop("context_data", None)
return await route_callback(message=message, back=False, **kwargs)

View File

@@ -0,0 +1,120 @@
from aiogram.types import InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from ..context import ContextData, CallbackCommand
def add_pagination_controls(
keyboard_builder: InlineKeyboardBuilder,
callback_data: ContextData,
total_pages: int,
command: CallbackCommand,
page: int,
):
if total_pages > 1:
navigation_buttons = []
ContextData(**callback_data.model_dump()).__setattr__
if total_pages > 10:
navigation_buttons.append(
InlineKeyboardButton(
text="⏮️",
callback_data=ContextData(
command=command,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data="1" if page != 1 else "skip",
).pack(),
)
)
navigation_buttons.append(
InlineKeyboardButton(
text="⏪️",
callback_data=ContextData(
command=command,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=str(max(page - 10, 1)) if page > 1 else "skip",
).pack(),
)
)
navigation_buttons.append(
InlineKeyboardButton(
text="◀️",
callback_data=ContextData(
command=command,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=str(max(page - 1, 1)) if page > 1 else "skip",
).pack(),
)
)
navigation_buttons.append(
InlineKeyboardButton(
text="▶️",
callback_data=ContextData(
command=command,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=(
str(min(page + 1, total_pages))
if page < total_pages
else "skip"
),
).pack(),
)
)
if total_pages > 10:
navigation_buttons.append(
InlineKeyboardButton(
text="⏩️",
callback_data=ContextData(
command=command,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=(
str(min(page + 10, total_pages))
if page < total_pages
else "skip"
),
).pack(),
)
)
navigation_buttons.append(
InlineKeyboardButton(
text="⏭️",
callback_data=ContextData(
command=command,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=str(total_pages) if page != total_pages else "skip",
).pack(),
)
)
keyboard_builder.row(*navigation_buttons)

View File

@@ -0,0 +1,59 @@
from aiogram.types import Message, CallbackQuery
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from qbot.main import QBotApp
from ..context import CallbackCommand
from ....utils.navigation import (
get_navigation_context,
save_navigation_context,
pop_navigation_context,
)
import qbot.bot.handlers.menu.main as menu_main
import qbot.bot.handlers.menu.settings as menu_settings
import qbot.bot.handlers.menu.parameters as menu_parameters
import qbot.bot.handlers.menu.language as menu_language
import qbot.bot.handlers.menu.entities as menu_entities
import qbot.bot.handlers.forms.entity_list as form_list
import qbot.bot.handlers.forms.entity_form as form_item
import qbot.bot.handlers.editors.main as editor
async def route_callback(message: Message | CallbackQuery, back: bool = True, **kwargs):
state_data = kwargs["state_data"]
stack, context = get_navigation_context(state_data)
if back:
context = pop_navigation_context(stack)
stack = save_navigation_context(callback_data=context, state_data=state_data)
kwargs.update({"callback_data": context, "navigation_stack": stack})
if context:
if context.command == CallbackCommand.MENU_ENTRY_MAIN:
await menu_main.main_menu(message, **kwargs)
elif context.command == CallbackCommand.MENU_ENTRY_SETTINGS:
await menu_settings.settings_menu(message, **kwargs)
elif context.command == CallbackCommand.MENU_ENTRY_PARAMETERS:
await menu_parameters.parameters_menu(message, **kwargs)
elif context.command == CallbackCommand.MENU_ENTRY_LANGUAGE:
await menu_language.language_menu(message, **kwargs)
elif context.command == CallbackCommand.MENU_ENTRY_ENTITIES:
await menu_entities.entities_menu(message, **kwargs)
elif context.command == CallbackCommand.ENTITY_LIST:
await form_list.entity_list(message, **kwargs)
elif context.command == CallbackCommand.ENTITY_ITEM:
await form_item.entity_item(message, **kwargs)
elif context.command == CallbackCommand.FIELD_EDITOR:
await editor.field_editor(message, **kwargs)
elif context.command == CallbackCommand.USER_COMMAND:
import qbot.bot.handlers.user_handlers.main as user_handler
app: "QBotApp" = kwargs["app"]
cmd = app.bot_commands.get(context.user_command.split("&")[0])
await user_handler.cammand_handler(message=message, cmd=cmd, **kwargs)
else:
raise ValueError(f"Unknown command {context.command}")
else:
raise ValueError("No navigation context")

View File

@@ -0,0 +1,44 @@
from aiogram.filters.callback_data import CallbackData as BaseCallbackData
from enum import StrEnum
class CallbackCommand(StrEnum):
FIELD_EDITOR = "fe"
FIELD_EDITOR_CALLBACK = "fc"
ENTITY_LIST = "el"
ENTITY_ITEM = "ei"
ENTITY_DELETE = "ed"
MENU_ENTRY_MAIN = "mm"
MENU_ENTRY_SETTINGS = "ms"
MENU_ENTRY_ENTITIES = "me"
MENU_ENTRY_PARAMETERS = "mp"
MENU_ENTRY_LANGUAGE = "ml"
SET_LANGUAGE = "ls"
DATE_PICKER_MONTH = "dm"
DATE_PICKER_YEAR = "dy"
TIME_PICKER = "tp"
# STRING_EDITOR_LOCALE = "sl"
ENTITY_PICKER_PAGE = "ep"
ENTITY_PICKER_TOGGLE_ITEM = "et"
VIEW_FILTER_EDIT = "vf"
USER_COMMAND = "uc"
class CommandContext(StrEnum):
SETTING_EDIT = "se"
ENTITY_CREATE = "ec"
ENTITY_EDIT = "ee"
ENTITY_FIELD_EDIT = "ef"
COMMAND_FORM = "cf"
class ContextData(BaseCallbackData, prefix="cd"):
command: CallbackCommand
context: CommandContext | None = None
entity_name: str | None = None
entity_id: int | None = None
field_name: str | None = None
form_params: str | None = None
user_command: str | None = None
data: str | None = None
back: bool = False

View File

@@ -0,0 +1,80 @@
from aiogram import Router
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from babel.support import LazyProxy
from logging import getLogger
from ....model.descriptors import FieldDescriptor
from ..context import ContextData, CallbackCommand
from ....utils.main import get_send_message
from .wrapper import wrap_editor
logger = getLogger(__name__)
router = Router()
async def bool_editor(
message: Message | CallbackQuery,
edit_prompt: str,
field_descriptor: FieldDescriptor,
callback_data: ContextData,
**kwargs,
):
keyboard_builder = InlineKeyboardBuilder()
if isinstance(field_descriptor.bool_true_value, LazyProxy):
true_caption = field_descriptor.bool_true_value.value
else:
true_caption = field_descriptor.bool_true_value
if isinstance(field_descriptor.bool_false_value, LazyProxy):
false_caption = field_descriptor.bool_false_value.value
else:
false_caption = field_descriptor.bool_false_value
keyboard_builder.row(
InlineKeyboardButton(
text=true_caption,
callback_data=ContextData(
command=CallbackCommand.FIELD_EDITOR_CALLBACK,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=str(True),
).pack(),
),
InlineKeyboardButton(
text=false_caption,
callback_data=ContextData(
command=CallbackCommand.FIELD_EDITOR_CALLBACK,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=str(False),
).pack(),
),
)
state_data = kwargs["state_data"]
await wrap_editor(
keyboard_builder=keyboard_builder,
field_descriptor=field_descriptor,
callback_data=callback_data,
state_data=state_data,
)
state: FSMContext = kwargs["state"]
await state.set_data(state_data)
send_message = get_send_message(message)
await send_message(text=edit_prompt, reply_markup=keyboard_builder.as_markup())

View File

@@ -0,0 +1,75 @@
from aiogram.types import Message, CallbackQuery
from decimal import Decimal
from datetime import datetime, time
from ....model.bot_entity import BotEntity
from ....model.bot_enum import BotEnum
from ....model.descriptors import FieldDescriptor
from ....model.settings import Settings
from ....model.user import UserBase
from ....utils.main import get_callable_str, get_value_repr
from ..context import ContextData, CommandContext
from .boolean import bool_editor
from .date import date_picker, time_picker
from .entity import entity_picker
from .string import string_editor
async def show_editor(message: Message | CallbackQuery, **kwargs):
field_descriptor: FieldDescriptor = kwargs["field_descriptor"]
current_value = kwargs["current_value"]
user: UserBase = kwargs["user"]
callback_data: ContextData = kwargs.get("callback_data", None)
state_data: dict = kwargs["state_data"]
value_type = field_descriptor.type_base
if field_descriptor.edit_prompt:
edit_prompt = get_callable_str(
field_descriptor.edit_prompt, field_descriptor, None, current_value
)
else:
if field_descriptor.caption:
caption_str = get_callable_str(
field_descriptor.caption, field_descriptor, None, current_value
)
else:
caption_str = field_descriptor.name
if callback_data.context == CommandContext.ENTITY_EDIT:
edit_prompt = (
await Settings.get(
Settings.APP_STRINGS_FIELD_EDIT_PROMPT_TEMPLATE_P_NAME_VALUE
)
).format(
name=caption_str,
value=get_value_repr(current_value, field_descriptor, user.lang),
)
else:
edit_prompt = (
await Settings.get(
Settings.APP_STRINGS_FIELD_CREATE_PROMPT_TEMPLATE_P_NAME
)
).format(name=caption_str)
kwargs["edit_prompt"] = edit_prompt
if value_type not in [int, float, Decimal, str]:
state_data.update({"context_data": callback_data.pack()})
if value_type is bool:
await bool_editor(message=message, **kwargs)
elif value_type in [int, float, Decimal, str]:
await string_editor(message=message, **kwargs)
elif value_type is datetime:
await date_picker(message=message, **kwargs)
elif value_type is time:
await time_picker(message=message, **kwargs)
elif issubclass(value_type, BotEntity) or issubclass(value_type, BotEnum):
await entity_picker(message=message, **kwargs)
else:
raise ValueError(f"Unsupported field type: {value_type}")

View File

@@ -0,0 +1,382 @@
from datetime import datetime, time, timedelta
from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from logging import getLogger
from typing import TYPE_CHECKING
from ....model.descriptors import FieldDescriptor
from ....model.settings import Settings
from ..context import ContextData, CallbackCommand
from ....utils.main import get_send_message, get_field_descriptor
from .wrapper import wrap_editor
if TYPE_CHECKING:
from ....main import QBotApp
logger = getLogger(__name__)
router = Router()
@router.callback_query(ContextData.filter(F.command == CallbackCommand.TIME_PICKER))
async def time_picker_callback(
query: CallbackQuery, callback_data: ContextData, app: "QBotApp", **kwargs
):
if not callback_data.data:
return
field_descriptor = get_field_descriptor(app, callback_data)
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
await time_picker(
query.message,
field_descriptor=field_descriptor,
callback_data=callback_data,
current_value=datetime.strptime(callback_data.data, "%Y-%m-%d %H-%M")
if len(callback_data.data) > 10
else time.fromisoformat(callback_data.data.replace("-", ":")),
**kwargs,
)
async def time_picker(
message: Message | CallbackQuery,
field_descriptor: FieldDescriptor,
callback_data: ContextData,
current_value: datetime | time,
state: FSMContext,
edit_prompt: str | None = None,
**kwargs,
):
keyboard_builder = InlineKeyboardBuilder()
if not current_value:
current_value = time(0, 0)
else:
remainder = current_value.minute % 5
if remainder >= 3:
current_value += timedelta(minutes=(5 - remainder))
else:
current_value -= timedelta(minutes=remainder)
for i in range(12):
keyboard_builder.row(
InlineKeyboardButton(
text=(
"▶︎ {v:02d} ◀︎" if i == (current_value.hour % 12) else "{v:02d}"
).format(v=i if current_value.hour < 12 else i + 12),
callback_data=ContextData(
command=CallbackCommand.TIME_PICKER,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=current_value.replace(
hour=i if current_value.hour < 12 else i + 12
).strftime(
"%Y-%m-%d %H-%M"
if isinstance(current_value, datetime)
else "%H-%M"
)
if i != current_value.hour % 12
else None,
).pack(),
),
InlineKeyboardButton(
text=(
"▶︎ {v:02d} ◀︎" if i == current_value.minute // 5 else "{v:02d}"
).format(v=i * 5),
callback_data=ContextData(
command=CallbackCommand.TIME_PICKER,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=current_value.replace(minute=i * 5).strftime(
"%Y-%m-%d %H-%M"
if isinstance(current_value, datetime)
else "%H-%M"
)
if i != current_value.minute // 5
else None,
).pack(),
),
)
keyboard_builder.row(
InlineKeyboardButton(
text="AM/PM",
callback_data=ContextData(
command=CallbackCommand.TIME_PICKER,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=current_value.replace(
hour=current_value.hour + 12
if current_value.hour < 12
else current_value.hour - 12
).strftime(
"%Y-%m-%d %H-%M" if isinstance(current_value, datetime) else "%H-%M"
),
).pack(),
),
InlineKeyboardButton(
text=await Settings.get(Settings.APP_STRINGS_DONE_BTN),
callback_data=ContextData(
command=CallbackCommand.FIELD_EDITOR_CALLBACK,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=current_value.strftime(
"%Y-%m-%d %H-%M" if isinstance(current_value, datetime) else "%H-%M"
),
).pack(),
),
)
state_data = kwargs["state_data"]
await wrap_editor(
keyboard_builder=keyboard_builder,
field_descriptor=field_descriptor,
callback_data=callback_data,
state_data=state_data,
)
await state.set_data(state_data)
if edit_prompt:
send_message = get_send_message(message)
await send_message(text=edit_prompt, reply_markup=keyboard_builder.as_markup())
else:
await message.edit_reply_markup(reply_markup=keyboard_builder.as_markup())
async def date_picker(
message: Message | CallbackQuery,
field_descriptor: FieldDescriptor,
callback_data: ContextData,
current_value: datetime,
state: FSMContext,
edit_prompt: str | None = None,
**kwargs,
):
if not current_value:
start_date = datetime.now()
else:
start_date = current_value
start_date = start_date.replace(day=1)
previous_month = start_date - timedelta(days=1)
next_month = start_date.replace(day=28) + timedelta(days=4)
keyboard_builder = InlineKeyboardBuilder()
keyboard_builder.row(
InlineKeyboardButton(
text="◀️",
callback_data=ContextData(
command=CallbackCommand.DATE_PICKER_MONTH,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=previous_month.strftime("%Y-%m-%d %H-%M"),
).pack(),
),
InlineKeyboardButton(
text=start_date.strftime("%b %Y"),
callback_data=ContextData(
command=CallbackCommand.DATE_PICKER_YEAR,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=start_date.strftime("%Y-%m-%d %H-%M"),
).pack(),
),
InlineKeyboardButton(
text="▶️",
callback_data=ContextData(
command=CallbackCommand.DATE_PICKER_MONTH,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=next_month.strftime("%Y-%m-%d %H-%M"),
).pack(),
),
)
first_day = start_date - timedelta(days=start_date.weekday())
weeks = (
(
(start_date.replace(day=28) + timedelta(days=4)).replace(day=1) - first_day
).days
- 1
) // 7 + 1
for week in range(weeks):
buttons = []
for day in range(7):
current_day = first_day + timedelta(days=week * 7 + day)
buttons.append(
InlineKeyboardButton(
text=current_day.strftime("%d"),
callback_data=ContextData(
command=CallbackCommand.FIELD_EDITOR_CALLBACK
if field_descriptor.dt_type == "date"
else CallbackCommand.TIME_PICKER,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=current_day.strftime("%Y-%m-%d %H-%M"),
).pack(),
)
)
keyboard_builder.row(*buttons)
state_data = kwargs["state_data"]
await wrap_editor(
keyboard_builder=keyboard_builder,
field_descriptor=field_descriptor,
callback_data=callback_data,
state_data=state_data,
)
await state.set_data(state_data)
if edit_prompt:
send_message = get_send_message(message)
await send_message(text=edit_prompt, reply_markup=keyboard_builder.as_markup())
else:
await message.edit_reply_markup(reply_markup=keyboard_builder.as_markup())
@router.callback_query(
ContextData.filter(F.command == CallbackCommand.DATE_PICKER_YEAR)
)
async def date_picker_year(
query: CallbackQuery,
callback_data: ContextData,
app: "QBotApp",
state: FSMContext,
**kwargs,
):
start_date = datetime.strptime(callback_data.data, "%Y-%m-%d %H-%M")
state_data = await state.get_data()
kwargs["state_data"] = state_data
keyboard_builder = InlineKeyboardBuilder()
keyboard_builder.row(
InlineKeyboardButton(
text="🔼",
callback_data=ContextData(
command=CallbackCommand.DATE_PICKER_YEAR,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=start_date.replace(year=start_date.year - 20).strftime(
"%Y-%m-%d %H-%M"
),
).pack(),
)
)
for r in range(4):
buttons = []
for c in range(5):
current_date = start_date.replace(year=start_date.year + r * 5 + c - 10)
buttons.append(
InlineKeyboardButton(
text=current_date.strftime("%Y"),
callback_data=ContextData(
command=CallbackCommand.DATE_PICKER_MONTH,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=current_date.strftime("%Y-%m-%d %H-%M"),
).pack(),
)
)
keyboard_builder.row(*buttons)
keyboard_builder.row(
InlineKeyboardButton(
text="🔽",
callback_data=ContextData(
command=CallbackCommand.DATE_PICKER_YEAR,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=start_date.replace(year=start_date.year + 20).strftime(
"%Y-%m-%d %H-%M"
),
).pack(),
)
)
field_descriptor = get_field_descriptor(app, callback_data)
await wrap_editor(
keyboard_builder=keyboard_builder,
field_descriptor=field_descriptor,
callback_data=callback_data,
state_data=state_data,
)
await query.message.edit_reply_markup(reply_markup=keyboard_builder.as_markup())
@router.callback_query(
ContextData.filter(F.command == CallbackCommand.DATE_PICKER_MONTH)
)
async def date_picker_month(
query: CallbackQuery, callback_data: ContextData, app: "QBotApp", **kwargs
):
field_descriptor = get_field_descriptor(app, callback_data)
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
await date_picker(
query.message,
field_descriptor=field_descriptor,
callback_data=callback_data,
current_value=datetime.strptime(callback_data.data, "%Y-%m-%d %H-%M"),
**kwargs,
)

View File

@@ -0,0 +1,362 @@
from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from logging import getLogger
from sqlmodel import column
from sqlmodel.ext.asyncio.session import AsyncSession
from typing import TYPE_CHECKING
from ....model.bot_entity import BotEntity
# from ....model.owned_bot_entity import OwnedBotEntity
from ....model.bot_enum import BotEnum
from ....model.settings import Settings
from ....model.user import UserBase
from ....model.view_setting import ViewSetting
from ....model.descriptors import FieldDescriptor, Filter
from ....model import EntityPermission
from ....utils.main import (
get_user_permissions,
get_send_message,
get_field_descriptor,
get_callable_str,
)
from ....utils.serialization import serialize, deserialize
from ..context import ContextData, CallbackCommand
from ..common.pagination import add_pagination_controls
from ..common.filtering import add_filter_controls
from .wrapper import wrap_editor
if TYPE_CHECKING:
from ....main import QBotApp
logger = getLogger(__name__)
router = Router()
async def entity_picker(
message: Message | CallbackQuery,
field_descriptor: FieldDescriptor,
edit_prompt: str,
current_value: BotEntity | BotEnum | list[BotEntity] | list[BotEnum],
**kwargs,
):
state_data: dict = kwargs["state_data"]
state_data.update(
{
"current_value": serialize(current_value, field_descriptor),
"value": serialize(current_value, field_descriptor),
"edit_prompt": edit_prompt,
}
)
await render_entity_picker(
field_descriptor=field_descriptor,
message=message,
current_value=current_value,
edit_prompt=edit_prompt,
**kwargs,
)
def calc_total_pages(items_count: int, page_size: int) -> int:
return max(items_count // page_size + (1 if items_count % page_size else 0), 1)
async def render_entity_picker(
*,
field_descriptor: FieldDescriptor,
message: Message | CallbackQuery,
callback_data: ContextData,
user: UserBase,
db_session: AsyncSession,
state: FSMContext,
current_value: BotEntity | BotEnum | list[BotEntity] | list[BotEnum],
edit_prompt: str,
page: int = 1,
**kwargs,
):
if callback_data.command in [
CallbackCommand.ENTITY_PICKER_PAGE,
CallbackCommand.ENTITY_PICKER_TOGGLE_ITEM,
]:
page = int(callback_data.data.split("&")[0])
type_ = field_descriptor.type_base
is_list = field_descriptor.is_list
if not issubclass(type_, BotEntity) and not issubclass(type_, BotEnum):
raise ValueError("Unsupported type")
page_size = await Settings.get(Settings.PAGE_SIZE)
form_list = None
if issubclass(type_, BotEnum):
items_count = len(type_.all_members)
total_pages = calc_total_pages(items_count, page_size)
page = min(page, total_pages)
enum_items = list(type_.all_members.values())[
page_size * (page - 1) : page_size * page
]
items = [
{
"text": f"{'' if not is_list else '【✔︎】 ' if item in (current_value or []) else '【 】 '}{item.localized(user.lang)}",
"value": item.value,
}
for item in enum_items
]
elif issubclass(type_, BotEntity):
form_name = field_descriptor.ep_form or "default"
form_list = type_.bot_entity_descriptor.lists.get(
form_name, type_.bot_entity_descriptor.default_list
)
permissions = get_user_permissions(user, type_.bot_entity_descriptor)
if form_list.filtering:
entity_filter = await ViewSetting.get_filter(
session=db_session,
user_id=user.id,
entity_name=type_.bot_entity_descriptor.class_name,
)
else:
entity_filter = None
list_all = EntityPermission.LIST_ALL in permissions
if list_all or EntityPermission.LIST in permissions:
if (
field_descriptor.ep_parent_field
and field_descriptor.ep_parent_field
and callback_data.entity_id
):
entity = await field_descriptor.entity_descriptor.type_.get(
session=db_session, id=callback_data.entity_id
)
value = getattr(entity, field_descriptor.ep_parent_field)
ext_filter = column(field_descriptor.ep_child_field).__eq__(value)
else:
ext_filter = None
if form_list.pagination:
items_count = await type_.get_count(
session=db_session,
static_filter=(
[
Filter(
field_name=f.field_name,
operator=f.operator,
value_type="const",
value=f.value,
)
for f in form_list.static_filters
if f.value_type == "const"
]
if isinstance(form_list.static_filters, list)
else form_list.static_filters
),
ext_filter=ext_filter,
filter=entity_filter,
filter_fields=form_list.filtering_fields,
user=user if not list_all else None,
)
total_pages = calc_total_pages(items_count, page_size)
page = min(page, total_pages)
skip = page_size * (page - 1)
limit = page_size
else:
skip = 0
limit = None
entity_items = await type_.get_multi(
session=db_session,
order_by=form_list.order_by,
static_filter=(
[
Filter(
field_name=f.field_name,
operator=f.operator,
value_type="const",
value=f.value,
)
for f in form_list.static_filters
if f.value_type == "const"
]
if isinstance(form_list.static_filters, list)
else form_list.static_filters
),
ext_filter=ext_filter,
filter=entity_filter,
user=user if not list_all else None,
skip=skip,
limit=limit,
)
else:
items_count = 0
total_pages = 1
page = 1
entity_items = list[BotEntity]()
items = [
{
"text": f"{
''
if not is_list
else '【✔︎】 '
if item in (current_value or [])
else '【 】 '
}{
type_.bot_entity_descriptor.item_repr(
type_.bot_entity_descriptor, item
)
if type_.bot_entity_descriptor.item_repr
else get_callable_str(
type_.bot_entity_descriptor.full_name,
type_.bot_entity_descriptor,
item,
)
if type_.bot_entity_descriptor.full_name
else f'{type_.bot_entity_descriptor.name}: {str(item.id)}'
}",
"value": str(item.id),
}
for item in entity_items
]
keyboard_builder = InlineKeyboardBuilder()
for item in items:
keyboard_builder.row(
InlineKeyboardButton(
text=item["text"],
callback_data=ContextData(
command=(
CallbackCommand.ENTITY_PICKER_TOGGLE_ITEM
if is_list
else CallbackCommand.FIELD_EDITOR_CALLBACK
),
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data=f"{page}&{item['value']}" if is_list else item["value"],
).pack(),
)
)
if form_list and form_list.pagination:
add_pagination_controls(
keyboard_builder=keyboard_builder,
callback_data=callback_data,
total_pages=total_pages,
command=CallbackCommand.ENTITY_PICKER_PAGE,
page=page,
)
if (
issubclass(type_, BotEntity)
and form_list.filtering
and form_list.filtering_fields
):
add_filter_controls(
keyboard_builder=keyboard_builder,
entity_descriptor=type_.bot_entity_descriptor,
filter=entity_filter,
filtering_fields=form_list.filtering_fields,
)
if is_list:
keyboard_builder.row(
InlineKeyboardButton(
text=await Settings.get(Settings.APP_STRINGS_DONE_BTN),
callback_data=ContextData(
command=CallbackCommand.FIELD_EDITOR_CALLBACK,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
).pack(),
)
)
state_data = kwargs["state_data"]
await wrap_editor(
keyboard_builder=keyboard_builder,
field_descriptor=field_descriptor,
callback_data=callback_data,
state_data=state_data,
)
await state.set_data(state_data)
send_message = get_send_message(message)
await send_message(text=edit_prompt, reply_markup=keyboard_builder.as_markup())
@router.callback_query(
ContextData.filter(F.command == CallbackCommand.ENTITY_PICKER_PAGE)
)
@router.callback_query(
ContextData.filter(F.command == CallbackCommand.ENTITY_PICKER_TOGGLE_ITEM)
)
async def entity_picker_callback(
query: CallbackQuery,
callback_data: ContextData,
db_session: AsyncSession,
app: "QBotApp",
state: FSMContext,
**kwargs,
):
state_data = await state.get_data()
kwargs["state_data"] = state_data
field_descriptor = get_field_descriptor(app=app, callback_data=callback_data)
# current_value = await deserialize(session = db_session, type_ = field_descriptor.type_, value = state_data["current_value"])
edit_prompt = state_data["edit_prompt"]
value = await deserialize(
session=db_session, type_=field_descriptor.type_, value=state_data["value"]
)
if callback_data.command == CallbackCommand.ENTITY_PICKER_TOGGLE_ITEM:
page, id_value = callback_data.data.split("&")
page = int(page)
type_ = field_descriptor.type_base
if issubclass(type_, BotEnum):
item = type_(id_value)
if item in value:
value.remove(item)
else:
value.append(item)
else:
item = await type_.get(session=db_session, id=int(id_value))
if item in value:
value.remove(item)
else:
value.append(item)
state_data.update({"value": serialize(value, field_descriptor)})
elif callback_data.command == CallbackCommand.ENTITY_PICKER_PAGE:
if callback_data.data == "skip":
return
page = int(callback_data.data)
else:
raise ValueError("Unsupported command")
await render_entity_picker(
field_descriptor=field_descriptor,
message=query,
callback_data=callback_data,
current_value=value,
edit_prompt=edit_prompt,
db_session=db_session,
app=app,
state=state,
page=page,
**kwargs,
)

View File

@@ -0,0 +1,164 @@
from typing import TYPE_CHECKING
from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery
from logging import getLogger
from sqlmodel.ext.asyncio.session import AsyncSession
from ....model import EntityPermission
from ....model.settings import Settings
from ....model.user import UserBase
from ....utils.main import (
check_entity_permission,
get_field_descriptor,
)
from ....utils.serialization import deserialize, serialize
from ..context import ContextData, CallbackCommand, CommandContext
from ....auth import authorize_command
from ....utils.navigation import (
get_navigation_context,
save_navigation_context,
)
from ..forms.entity_form import entity_item
from .common import show_editor
from ..menu.parameters import parameters_menu
from .string import router as string_editor_router
from .date import router as date_picker_router
from .boolean import router as bool_editor_router
from .entity import router as entity_picker_router
if TYPE_CHECKING:
from ....main import QBotApp
logger = getLogger(__name__)
router = Router()
@router.callback_query(ContextData.filter(F.command == CallbackCommand.FIELD_EDITOR))
async def field_editor_callback(query: CallbackQuery, **kwargs):
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
await field_editor(message=query, **kwargs)
async def field_editor(message: Message | CallbackQuery, **kwargs):
callback_data: ContextData = kwargs.get("callback_data", None)
db_session: AsyncSession = kwargs["db_session"]
user: UserBase = kwargs["user"]
app: "QBotApp" = kwargs["app"]
# state: FSMContext = kwargs["state"]
state_data: dict = kwargs["state_data"]
entity_data = state_data.get("entity_data")
for key in ["current_value", "value", "locale_index"]:
if key in state_data:
state_data.pop(key)
kwargs["state_data"] = state_data
entity_descriptor = None
if callback_data.context == CommandContext.SETTING_EDIT:
field_descriptor = get_field_descriptor(app, callback_data)
if field_descriptor.type_ is bool:
if await authorize_command(user=user, callback_data=callback_data):
await Settings.set_param(
field_descriptor, not await Settings.get(field_descriptor)
)
else:
return await message.answer(
text=(await Settings.get(Settings.APP_STRINGS_FORBIDDEN))
)
stack, context = get_navigation_context(state_data=state_data)
return await parameters_menu(
message=message, navigation_stack=stack, **kwargs
)
current_value = await Settings.get(field_descriptor, all_locales=True)
else:
field_descriptor = get_field_descriptor(app, callback_data)
entity_descriptor = field_descriptor.entity_descriptor
current_value = None
if (
field_descriptor.type_base is bool
and callback_data.context == CommandContext.ENTITY_FIELD_EDIT
):
entity = await entity_descriptor.type_.get(
session=db_session, id=int(callback_data.entity_id)
)
if check_entity_permission(
entity=entity, user=user, permission=EntityPermission.UPDATE
):
current_value: bool = (
getattr(entity, field_descriptor.field_name) or False
)
setattr(entity, field_descriptor.field_name, not current_value)
await db_session.commit()
stack, context = get_navigation_context(state_data=state_data)
return await entity_item(
query=message, navigation_stack=stack, **kwargs
)
if not entity_data and callback_data.context in [
CommandContext.ENTITY_EDIT,
CommandContext.ENTITY_FIELD_EDIT,
]:
entity = await entity_descriptor.type_.get(
session=kwargs["db_session"], id=int(callback_data.entity_id)
)
if check_entity_permission(
entity=entity, user=user, permission=EntityPermission.READ
):
if entity:
form_name = (
callback_data.form_params.split("&")[0]
if callback_data.form_params
else "default"
)
form = entity_descriptor.forms.get(
form_name, entity_descriptor.default_form
)
entity_data = {
key: serialize(
getattr(entity, key),
entity_descriptor.fields_descriptors[key],
)
for key in (
form.edit_field_sequence
if callback_data.context == CommandContext.ENTITY_EDIT
else [callback_data.field_name]
)
}
state_data.update({"entity_data": entity_data})
if entity_data:
current_value = await deserialize(
session=db_session,
type_=field_descriptor.type_,
value=entity_data.get(callback_data.field_name),
)
kwargs.update({"field_descriptor": field_descriptor})
save_navigation_context(state_data=state_data, callback_data=callback_data)
await show_editor(message=message, current_value=current_value, **kwargs)
router.include_routers(
string_editor_router,
date_picker_router,
bool_editor_router,
entity_picker_router,
)

View File

@@ -0,0 +1,349 @@
from aiogram import Router, F
from aiogram.types import Message, CallbackQuery
from aiogram.fsm.context import FSMContext
from sqlmodel.ext.asyncio.session import AsyncSession
from typing import TYPE_CHECKING
from decimal import Decimal
import json
from ..context import ContextData, CallbackCommand, CommandContext
from ...command_context_filter import CallbackCommandFilter
from ..user_handlers.main import cammand_handler
from ....model import EntityPermission
from ....model.user import UserBase
from ....model.settings import Settings
from ....model.descriptors import FieldDescriptor
from ....model.language import LanguageBase
from ....auth import authorize_command
from ....utils.main import (
get_user_permissions,
check_entity_permission,
clear_state,
get_entity_descriptor,
get_field_descriptor,
)
from ....utils.serialization import deserialize
from ..common.routing import route_callback
from .common import show_editor
if TYPE_CHECKING:
from ....main import QBotApp
router = Router()
@router.message(CallbackCommandFilter(CallbackCommand.FIELD_EDITOR_CALLBACK))
@router.callback_query(
ContextData.filter(F.command == CallbackCommand.FIELD_EDITOR_CALLBACK)
)
async def field_editor_callback(message: Message | CallbackQuery, **kwargs):
app: "QBotApp" = kwargs["app"]
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
if isinstance(message, Message):
callback_data: ContextData = kwargs.get("callback_data", None)
context_data = state_data.get("context_data")
if context_data:
callback_data = ContextData.unpack(context_data)
value = message.text
field_descriptor = get_field_descriptor(app, callback_data)
type_base = field_descriptor.type_base
if type_base is str and field_descriptor.localizable:
locale_index = int(state_data.get("locale_index"))
value = state_data.get("value")
if value:
value = json.loads(value)
else:
value = {}
value[list(LanguageBase.all_members.values())[locale_index].value] = (
message.text
)
value = json.dumps(value, ensure_ascii=False)
if locale_index < len(LanguageBase.all_members.values()) - 1:
current_value = state_data.get("current_value")
state_data.update({"value": value})
entity_descriptor = get_entity_descriptor(app, callback_data)
kwargs.update({"callback_data": callback_data})
return await show_editor(
message=message,
locale_index=locale_index + 1,
field_descriptor=field_descriptor,
entity_descriptor=entity_descriptor,
current_value=current_value,
value=value,
**kwargs,
)
# else:
# value = state_data.get("value")
# if value:
# value = json.loads(value)
# else:
# value = {}
# value[list(LanguageBase.all_members.keys())[locale_index]] = (
# message.text
# )
# value = json.dumps(value, ensure_ascii=False)
elif type_base in [int, float, Decimal]:
try:
_ = type_base(value) # @IgnoreException
except Exception:
return await message.answer(
text=(await Settings.get(Settings.APP_STRINGS_INVALID_INPUT))
)
else:
callback_data: ContextData = kwargs["callback_data"]
if callback_data.data:
if callback_data.data == "skip":
value = None
else:
value = callback_data.data
else:
value = state_data.get("value")
field_descriptor = get_field_descriptor(app, callback_data)
kwargs.update(
{
"callback_data": callback_data,
}
)
await process_field_edit_callback(
message=message, value=value, field_descriptor=field_descriptor, **kwargs
)
async def process_field_edit_callback(message: Message | CallbackQuery, **kwargs):
user: UserBase = kwargs["user"]
db_session: AsyncSession = kwargs["db_session"]
callback_data: ContextData = kwargs.get("callback_data", None)
state_data: dict = kwargs["state_data"]
value = kwargs["value"]
field_descriptor: FieldDescriptor = kwargs["field_descriptor"]
if callback_data.context == CommandContext.SETTING_EDIT:
if callback_data.data != "cancel":
if await authorize_command(user=user, callback_data=callback_data):
value = await deserialize(
session=db_session, type_=field_descriptor.type_, value=value
)
await Settings.set_param(field_descriptor, value)
else:
return await message.answer(
text=(await Settings.get(Settings.APP_STRINGS_FORBIDDEN))
)
return await route_callback(message=message, back=True, **kwargs)
elif callback_data.context in [
CommandContext.ENTITY_CREATE,
CommandContext.ENTITY_EDIT,
CommandContext.ENTITY_FIELD_EDIT,
CommandContext.COMMAND_FORM,
]:
app: "QBotApp" = kwargs["app"]
entity_descriptor = get_entity_descriptor(app, callback_data)
if callback_data.context == CommandContext.COMMAND_FORM:
field_sequence = list(field_descriptor.command.param_form.keys())
current_index = field_sequence.index(callback_data.field_name)
field_descriptors = field_descriptor.command.param_form
else:
form_name = (
callback_data.form_params.split("&")[0]
if callback_data.form_params
else "default"
)
form = entity_descriptor.forms.get(
form_name, entity_descriptor.default_form
)
field_sequence = form.edit_field_sequence
current_index = (
field_sequence.index(callback_data.field_name)
if callback_data.context
in [CommandContext.ENTITY_CREATE, CommandContext.ENTITY_EDIT]
else 0
)
field_descriptors = entity_descriptor.fields_descriptors
entity_data = state_data.get("entity_data", {})
if callback_data.context == CommandContext.ENTITY_CREATE and not entity_data:
stack = state_data.get("navigation_stack", [])
prev_callback_data = ContextData.unpack(stack[-1]) if stack else None
if (
prev_callback_data
and prev_callback_data.command == CallbackCommand.ENTITY_LIST
and prev_callback_data.entity_name == entity_descriptor.name
):
prev_form_name = (
prev_callback_data.form_params.split("&")[0]
if prev_callback_data.form_params
else "default"
)
prev_form_params = (
prev_callback_data.form_params.split("&")[1:]
if prev_callback_data.form_params
else []
)
prev_form_list = entity_descriptor.lists.get(
prev_form_name or "default", entity_descriptor.default_list
)
if prev_form_list.static_filters:
for filt in prev_form_list.static_filters:
if filt.value_type == "const":
entity_data[filt.field_name] = filt.value
elif len(prev_form_params) > filt.param_index:
entity_data[filt.field_name] = prev_form_params[
filt.param_index
]
if (
callback_data.context
in [
CommandContext.ENTITY_CREATE,
CommandContext.ENTITY_EDIT,
CommandContext.COMMAND_FORM,
]
and current_index < len(field_sequence) - 1
):
entity_data[field_descriptor.field_name] = value
state_data.update({"entity_data": entity_data})
next_field_name = field_sequence[current_index + 1]
next_field_descriptor = field_descriptors[next_field_name]
kwargs.update({"field_descriptor": next_field_descriptor})
callback_data.field_name = next_field_name
state_entity_val = entity_data.get(next_field_descriptor.field_name)
current_value = (
await deserialize(
session=db_session,
type_=next_field_descriptor.type_,
value=state_entity_val,
)
if state_entity_val
else None
)
await show_editor(
message=message,
entity_descriptor=entity_descriptor,
current_value=current_value,
**kwargs,
)
else:
entity_data[field_descriptor.field_name] = value
# What if user has several roles and each role has its own ownership field? Should we allow creation even
# if user has no CREATE_ALL permission
# for role in user.roles:
# if role in entity_descriptor.ownership_fields and not EntityPermission.CREATE_ALL in user_permissions:
# entity_data[entity_descriptor.ownership_fields[role]] = user.id
deser_entity_data = {
key: await deserialize(
session=db_session,
type_=field_descriptors[key].type_,
value=value,
)
for key, value in entity_data.items()
}
if callback_data.context == CommandContext.ENTITY_CREATE:
entity_type = entity_descriptor.type_
user_permissions = get_user_permissions(user, entity_descriptor)
if (
EntityPermission.CREATE not in user_permissions
and EntityPermission.CREATE_ALL not in user_permissions
):
return await message.answer(
text=(await Settings.get(Settings.APP_STRINGS_FORBIDDEN))
)
new_entity = await entity_type.create(
session=db_session,
obj_in=entity_type(**deser_entity_data),
commit=True,
)
form_name = (
callback_data.form_params.split("&")[0]
if callback_data.form_params
else "default"
)
form_list = entity_descriptor.lists.get(
form_name or "default", entity_descriptor.default_list
)
state_data["navigation_context"] = ContextData(
command=CallbackCommand.ENTITY_ITEM,
entity_name=entity_descriptor.name,
form_params=form_list.item_form,
entity_id=str(new_entity.id),
).pack()
state_data.update(state_data)
clear_state(state_data=state_data)
return await route_callback(message=message, back=False, **kwargs)
elif callback_data.context in [
CommandContext.ENTITY_EDIT,
CommandContext.ENTITY_FIELD_EDIT,
]:
entity_type = entity_descriptor.type_
entity_id = int(callback_data.entity_id)
entity = await entity_type.get(session=db_session, id=entity_id)
if not entity:
return await message.answer(
text=(await Settings.get(Settings.APP_STRINGS_NOT_FOUND))
)
if not check_entity_permission(
entity=entity, user=user, permission=EntityPermission.UPDATE
):
return await message.answer(
text=(await Settings.get(Settings.APP_STRINGS_FORBIDDEN))
)
for key, value in deser_entity_data.items():
setattr(entity, key, value)
await db_session.commit()
elif callback_data.context == CommandContext.COMMAND_FORM:
clear_state(state_data=state_data)
state_data["entity_data"] = entity_data
kwargs.update(
{
"callback_data": ContextData(
command=CallbackCommand.USER_COMMAND,
user_command=callback_data.user_command,
)
}
)
cmd = app.bot_commands.get(callback_data.user_command.split("&")[0])
return await cammand_handler(message=message, cmd=cmd, **kwargs)
clear_state(state_data=state_data)
# TODO: Try back=False and check if it works to navigate to newly created entity
await route_callback(message=message, back=True, **kwargs)

View File

@@ -0,0 +1,98 @@
from aiogram import Router
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton, CopyTextButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from logging import getLogger
from typing import Any
from ....model.descriptors import FieldDescriptor
from ....model.language import LanguageBase
from ....model.settings import Settings
from ....utils.main import get_send_message, get_local_text
from ....utils.serialization import serialize
from ..context import ContextData, CallbackCommand
from .wrapper import wrap_editor
logger = getLogger(__name__)
router = Router()
async def string_editor(
message: Message | CallbackQuery,
field_descriptor: FieldDescriptor,
callback_data: ContextData,
current_value: Any,
edit_prompt: str,
state: FSMContext,
locale_index: int = 0,
**kwargs,
):
keyboard_builder = InlineKeyboardBuilder()
state_data: dict = kwargs["state_data"]
_edit_prompt = edit_prompt
context_data = ContextData(
command=CallbackCommand.FIELD_EDITOR_CALLBACK,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
)
if field_descriptor.type_base is str and field_descriptor.localizable:
current_locale = list(LanguageBase.all_members.values())[locale_index]
_edit_prompt = f"{edit_prompt}\n{
(
await Settings.get(
Settings.APP_STRINGS_STRING_EDITOR_LOCALE_TEMPLATE_P_NAME
)
).format(name=current_locale)
}"
_current_value = (
get_local_text(current_value, current_locale) if current_value else None
)
state_data.update(
{
"context_data": context_data.pack(),
"edit_prompt": edit_prompt,
"locale_index": str(locale_index),
"current_value": current_value,
}
)
else:
_current_value = serialize(current_value, field_descriptor)
state_data.update({"context_data": context_data.pack()})
if _current_value:
_current_value_caption = (
f"{_current_value[:30]}..." if len(_current_value) > 30 else _current_value
)
keyboard_builder.row(
InlineKeyboardButton(
text=_current_value_caption,
copy_text=CopyTextButton(text=_current_value),
)
)
state_data = kwargs["state_data"]
await wrap_editor(
keyboard_builder=keyboard_builder,
field_descriptor=field_descriptor,
callback_data=callback_data,
state_data=state_data,
)
await state.set_data(state_data)
send_message = get_send_message(message)
await send_message(text=_edit_prompt, reply_markup=keyboard_builder.as_markup())

View File

@@ -0,0 +1,105 @@
from aiogram.types import InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from ....model.settings import Settings
from ....model.descriptors import FieldDescriptor
from ..context import ContextData, CallbackCommand, CommandContext
from ....utils.navigation import get_navigation_context, pop_navigation_context
async def wrap_editor(
keyboard_builder: InlineKeyboardBuilder,
field_descriptor: FieldDescriptor,
callback_data: ContextData,
state_data: dict,
):
if callback_data.context in [
CommandContext.ENTITY_CREATE,
CommandContext.ENTITY_EDIT,
CommandContext.ENTITY_FIELD_EDIT,
CommandContext.COMMAND_FORM,
]:
btns = []
show_back = True
show_cancel = True
if callback_data.context == CommandContext.COMMAND_FORM:
field_sequence = list(field_descriptor.command.param_form.keys())
field_index = field_sequence.index(callback_data.field_name)
show_back = field_descriptor.command.show_back_in_param_form
show_cancel = field_descriptor.command.show_cancel_in_param_form
else:
form_name = (
callback_data.form_params.split("&")[0]
if callback_data.form_params
else "default"
)
form = field_descriptor.entity_descriptor.forms.get(
form_name, field_descriptor.entity_descriptor.default_form
)
field_sequence = form.edit_field_sequence
field_index = (
field_sequence.index(field_descriptor.name)
if callback_data.context
in [CommandContext.ENTITY_CREATE, CommandContext.ENTITY_EDIT]
else 0
)
stack, context = get_navigation_context(state_data=state_data)
context = pop_navigation_context(stack)
if field_index > 0 and show_back:
btns.append(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_BACK_BTN)),
callback_data=ContextData(
command=CallbackCommand.FIELD_EDITOR,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
field_name=field_sequence[field_index - 1],
).pack(),
)
)
if field_descriptor.is_optional:
btns.append(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_SKIP_BTN)),
callback_data=ContextData(
command=CallbackCommand.FIELD_EDITOR_CALLBACK,
context=callback_data.context,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
user_command=callback_data.user_command,
data="skip",
).pack(),
)
)
keyboard_builder.row(*btns)
if show_cancel:
keyboard_builder.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_CANCEL_BTN)),
callback_data=context.pack(),
)
)
elif callback_data.context == CommandContext.SETTING_EDIT:
keyboard_builder.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_CANCEL_BTN)),
callback_data=ContextData(
command=CallbackCommand.FIELD_EDITOR_CALLBACK,
context=callback_data.context,
field_name=callback_data.field_name,
form_params=callback_data.form_params,
data="cancel",
).pack(),
)
)

View File

@@ -0,0 +1,270 @@
from typing import TYPE_CHECKING
from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.types import CallbackQuery, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from logging import getLogger
from sqlmodel.ext.asyncio.session import AsyncSession
from ....model.descriptors import FieldEditButton, CommandButton, InlineButton
from ....model.settings import Settings
from ....model.user import UserBase
from ....model import EntityPermission
from ....utils.main import (
check_entity_permission,
get_send_message,
clear_state,
get_value_repr,
get_callable_str,
get_entity_descriptor,
)
from ..context import ContextData, CallbackCommand, CommandContext
from ....utils.navigation import (
pop_navigation_context,
save_navigation_context,
)
if TYPE_CHECKING:
from ....main import QBotApp
logger = getLogger(__name__)
router = Router()
@router.callback_query(ContextData.filter(F.command == CallbackCommand.ENTITY_ITEM))
async def entity_item_callback(query: CallbackQuery, **kwargs):
callback_data: ContextData = kwargs["callback_data"]
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
clear_state(state_data=state_data)
stack = save_navigation_context(callback_data=callback_data, state_data=state_data)
await entity_item(query=query, navigation_stack=stack, **kwargs)
async def entity_item(
query: CallbackQuery,
callback_data: ContextData,
db_session: AsyncSession,
user: UserBase,
app: "QBotApp",
navigation_stack: list[ContextData],
**kwargs,
):
entity_descriptor = get_entity_descriptor(app, callback_data)
# user_permissions = get_user_permissions(user, entity_descriptor)
entity_type = entity_descriptor.type_
keyboard_builder = InlineKeyboardBuilder()
entity_item = await entity_type.get(session=db_session, id=callback_data.entity_id)
if not entity_item:
return await query.answer(
text=(await Settings.get(Settings.APP_STRINGS_NOT_FOUND))
)
# is_owned = issubclass(entity_type, OwnedBotEntity)
if not check_entity_permission(
entity=entity_item, user=user, permission=EntityPermission.READ
):
return await query.answer(
text=(await Settings.get(Settings.APP_STRINGS_FORBIDDEN))
)
can_edit = check_entity_permission(
entity=entity_item, user=user, permission=EntityPermission.UPDATE
)
form = entity_descriptor.forms.get(
callback_data.form_params or "default", entity_descriptor.default_form
)
if form.form_buttons:
for edit_buttons_row in form.form_buttons:
btn_row = []
for button in edit_buttons_row:
if button.visibility and not button.visibility(entity_item):
continue
if isinstance(button, FieldEditButton) and can_edit:
field_name = button.field_name
btn_caption = button.caption
if field_name in entity_descriptor.fields_descriptors:
field_descriptor = entity_descriptor.fields_descriptors[
field_name
]
field_value = getattr(entity_item, field_descriptor.field_name)
if btn_caption:
btn_text = get_callable_str(
btn_caption, field_descriptor, entity_item, field_value
)
else:
if field_descriptor.type_base is bool:
btn_text = f"{'【✔︎】 ' if field_value else '【 】 '}{
get_callable_str(
field_descriptor.caption,
field_descriptor,
entity_item,
field_value,
)
if field_descriptor.caption
else field_name
}"
else:
btn_text = (
f"✏️ {
get_callable_str(
field_descriptor.caption,
field_descriptor,
entity_item,
field_value,
)
}"
if field_descriptor.caption
else f"✏️ {field_name}"
)
btn_row.append(
InlineKeyboardButton(
text=btn_text,
callback_data=ContextData(
command=CallbackCommand.FIELD_EDITOR,
context=CommandContext.ENTITY_FIELD_EDIT,
entity_name=entity_descriptor.name,
entity_id=str(entity_item.id),
field_name=field_name,
).pack(),
)
)
elif isinstance(button, CommandButton):
btn_caption = button.caption
btn_text = get_callable_str(
btn_caption, entity_descriptor, entity_item
)
if isinstance(button.command, ContextData):
btn_cdata = button.command
elif callable(button.command):
btn_cdata = button.command(callback_data, entity_item)
elif isinstance(button.command, str):
btn_cdata = ContextData(
command=CallbackCommand.USER_COMMAND,
user_command=button.command,
)
btn_row.append(
InlineKeyboardButton(
text=btn_text,
callback_data=btn_cdata.pack(),
)
)
elif isinstance(button, InlineButton):
if isinstance(button.inline_button, InlineKeyboardButton):
btn_row.append(button.inline_button)
elif callable(button.inline_button):
btn_row.append(button.inline_button(entity_item))
if btn_row:
keyboard_builder.row(*btn_row)
edit_delete_row = []
if can_edit and form.show_edit_button:
edit_delete_row.append(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_EDIT_BTN)),
callback_data=ContextData(
command=CallbackCommand.FIELD_EDITOR,
context=CommandContext.ENTITY_EDIT,
entity_name=entity_descriptor.name,
entity_id=str(entity_item.id),
form_params=callback_data.form_params,
field_name=form.edit_field_sequence[0],
).pack(),
)
)
if (
check_entity_permission(
entity=entity_item, user=user, permission=EntityPermission.DELETE
)
and form.show_delete_button
):
edit_delete_row.append(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_DELETE_BTN)),
callback_data=ContextData(
command=CallbackCommand.ENTITY_DELETE,
entity_name=entity_descriptor.name,
form_params=callback_data.form_params,
entity_id=str(entity_item.id),
).pack(),
)
)
if edit_delete_row:
keyboard_builder.row(*edit_delete_row)
if form.item_repr:
item_text = form.item_repr(entity_descriptor, entity_item)
else:
entity_caption = (
get_callable_str(
entity_descriptor.full_name, entity_descriptor, entity_item
)
if entity_descriptor.full_name
else entity_descriptor.name
)
entity_item_repr = (
get_callable_str(
entity_descriptor.item_repr, entity_descriptor, entity_item
)
if entity_descriptor.item_repr
else str(entity_item.id)
)
item_text = f"<b><u><i>{entity_caption or entity_descriptor.name}:</i></u></b> <b>{entity_item_repr}</b>"
for field_descriptor in entity_descriptor.fields_descriptors.values():
if field_descriptor.is_visible:
field_caption = get_callable_str(
field_descriptor.caption, field_descriptor, entity_item
)
if field_descriptor.caption_value:
value = get_callable_str(
field_descriptor.caption_value,
field_descriptor,
entity_item,
getattr(entity_item, field_descriptor.field_name),
)
else:
value = get_value_repr(
value=getattr(entity_item, field_descriptor.field_name),
field_descriptor=field_descriptor,
locale=user.lang,
)
item_text += f"\n{field_caption or field_descriptor.name}:{f' <b>{value}</b>' if value else ''}"
context = pop_navigation_context(navigation_stack)
if context:
keyboard_builder.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_BACK_BTN)),
callback_data=context.pack(),
)
)
state: FSMContext = kwargs["state"]
state_data = kwargs["state_data"]
await state.set_data(state_data)
send_message = get_send_message(query)
await send_message(text=item_text, reply_markup=keyboard_builder.as_markup())

View File

@@ -0,0 +1,88 @@
from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.types import CallbackQuery, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from sqlmodel.ext.asyncio.session import AsyncSession
from typing import TYPE_CHECKING
from ..context import ContextData, CallbackCommand
from ....model.user import UserBase
from ....model.settings import Settings
from ....model import EntityPermission
from ....utils.main import (
check_entity_permission,
get_entity_item_repr,
get_entity_descriptor,
)
from ..common.routing import route_callback
if TYPE_CHECKING:
from ....main import QBotApp
router = Router()
@router.callback_query(ContextData.filter(F.command == CallbackCommand.ENTITY_DELETE))
async def entity_delete_callback(query: CallbackQuery, **kwargs):
callback_data: ContextData = kwargs["callback_data"]
user: UserBase = kwargs["user"]
db_session: AsyncSession = kwargs["db_session"]
app: "QBotApp" = kwargs["app"]
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
entity_descriptor = get_entity_descriptor(app=app, callback_data=callback_data)
entity = await entity_descriptor.type_.get(
session=db_session, id=int(callback_data.entity_id)
)
if not check_entity_permission(
entity=entity, user=user, permission=EntityPermission.DELETE
):
return await query.answer(
text=(await Settings.get(Settings.APP_STRINGS_FORBIDDEN))
)
if callback_data.data == "yes":
await entity_descriptor.type_.remove(
session=db_session, id=int(callback_data.entity_id), commit=True
)
await route_callback(message=query, **kwargs)
elif not callback_data.data:
entity = await entity_descriptor.type_.get(
session=db_session, id=int(callback_data.entity_id)
)
return await query.message.edit_text(
text=(
await Settings.get(Settings.APP_STRINGS_CONFIRM_DELETE_P_NAME)
).format(name=get_entity_item_repr(entity=entity)),
reply_markup=InlineKeyboardBuilder()
.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_YES_BTN)),
callback_data=ContextData(
command=CallbackCommand.ENTITY_DELETE,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
form_params=callback_data.form_params,
data="yes",
).pack(),
),
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_NO_BTN)),
callback_data=ContextData(
command=CallbackCommand.ENTITY_ITEM,
entity_name=callback_data.entity_name,
entity_id=callback_data.entity_id,
form_params=callback_data.form_params,
).pack(),
),
)
.as_markup(),
)

View File

@@ -0,0 +1,266 @@
from typing import TYPE_CHECKING
from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from logging import getLogger
from sqlmodel.ext.asyncio.session import AsyncSession
from ....model.bot_entity import BotEntity
from ....model.settings import Settings
from ....model.user import UserBase
from ....model.view_setting import ViewSetting
from ....model.descriptors import EntityDescriptor, Filter
from ....model import EntityPermission
from ....utils.main import (
get_user_permissions,
get_send_message,
clear_state,
get_entity_descriptor,
get_callable_str,
)
from ....utils.serialization import deserialize
from ..context import ContextData, CallbackCommand, CommandContext
from ..common.pagination import add_pagination_controls
from ..common.filtering import add_filter_controls
from ....utils.navigation import pop_navigation_context, save_navigation_context
if TYPE_CHECKING:
from ....main import QBotApp
logger = getLogger(__name__)
router = Router()
@router.callback_query(ContextData.filter(F.command == CallbackCommand.ENTITY_LIST))
async def entity_list_callback(query: CallbackQuery, **kwargs):
callback_data: ContextData = kwargs["callback_data"]
if callback_data.data == "skip":
return
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
clear_state(state_data=state_data)
stack = save_navigation_context(callback_data=callback_data, state_data=state_data)
await entity_list(message=query, navigation_stack=stack, **kwargs)
def calc_total_pages(items_count: int, page_size: int) -> int:
return max(items_count // page_size + (1 if items_count % page_size else 0), 1)
async def _prepare_static_filter(
db_session: AsyncSession,
entity_descriptor: EntityDescriptor,
static_filters: list[Filter],
params: list[str],
) -> list[Filter]:
return (
[
Filter(
field_name=f.field_name,
operator=f.operator,
value_type="const",
value=(
f.value
if f.value_type == "const"
else await deserialize(
session=db_session,
type_=entity_descriptor.fields_descriptors[
f.field_name
].type_base,
value=params[f.param_index],
)
),
)
for f in static_filters
]
if static_filters
else None
)
async def entity_list(
message: CallbackQuery | Message,
callback_data: ContextData,
db_session: AsyncSession,
user: UserBase,
app: "QBotApp",
navigation_stack: list[ContextData],
**kwargs,
):
page = int(callback_data.data or "1")
entity_descriptor = get_entity_descriptor(app, callback_data)
user_permissions = get_user_permissions(user, entity_descriptor)
entity_type = entity_descriptor.type_
form_params = (
callback_data.form_params.split("&") if callback_data.form_params else []
)
form_name = form_params.pop(0) if form_params else "default"
form_list = entity_descriptor.lists.get(
form_name or "default", entity_descriptor.default_list
)
form_item = entity_descriptor.forms.get(
form_list.item_form or "default", entity_descriptor.default_form
)
keyboard_builder = InlineKeyboardBuilder()
if (
EntityPermission.CREATE in user_permissions
or EntityPermission.CREATE_ALL in user_permissions
) and form_list.show_add_new_button:
keyboard_builder.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_ADD_BTN)),
callback_data=ContextData(
command=CallbackCommand.FIELD_EDITOR,
context=CommandContext.ENTITY_CREATE,
entity_name=entity_descriptor.name,
field_name=form_item.edit_field_sequence[0],
form_params=form_list.item_form,
).pack(),
)
)
if form_list.filtering:
entity_filter = await ViewSetting.get_filter(
session=db_session,
user_id=user.id,
entity_name=entity_descriptor.class_name,
)
else:
entity_filter = None
list_all = (
EntityPermission.LIST_ALL in user_permissions
or EntityPermission.READ_ALL in user_permissions
)
if (
list_all
or EntityPermission.LIST in user_permissions
or EntityPermission.READ in user_permissions
):
if form_list.pagination:
page_size = await Settings.get(Settings.PAGE_SIZE)
items_count = await entity_type.get_count(
session=db_session,
static_filter=await _prepare_static_filter(
db_session=db_session,
entity_descriptor=entity_descriptor,
static_filters=form_list.static_filters,
params=form_params,
),
filter=entity_filter,
filter_fields=form_list.filtering_fields,
user=user if not list_all else None,
)
total_pages = calc_total_pages(items_count, page_size)
page = min(page, total_pages)
skip = page_size * (page - 1)
limit = page_size
else:
skip = 0
limit = None
items = await entity_type.get_multi(
session=db_session,
order_by=form_list.order_by,
static_filter=await _prepare_static_filter(
db_session=db_session,
entity_descriptor=entity_descriptor,
static_filters=form_list.static_filters,
params=form_params,
),
filter=entity_filter,
filter_fields=form_list.filtering_fields,
user=user if not list_all else None,
skip=skip,
limit=limit,
)
else:
items = list[BotEntity]()
items_count = 0
total_pages = 1
page = 1
for item in items:
if form_list.item_repr:
caption = form_list.item_repr(entity_descriptor, item)
elif entity_descriptor.item_repr:
caption = entity_descriptor.item_repr(entity_descriptor, item)
elif entity_descriptor.full_name:
caption = f"{
get_callable_str(
callable_str=entity_descriptor.full_name,
descriptor=entity_descriptor,
entity=item,
)
}: {item.id}"
else:
caption = f"{entity_descriptor.name}: {item.id}"
keyboard_builder.row(
InlineKeyboardButton(
text=caption,
callback_data=ContextData(
command=CallbackCommand.ENTITY_ITEM,
entity_name=entity_descriptor.name,
form_params=form_list.item_form,
entity_id=str(item.id),
).pack(),
)
)
if form_list.pagination:
add_pagination_controls(
keyboard_builder=keyboard_builder,
callback_data=callback_data,
total_pages=total_pages,
command=CallbackCommand.ENTITY_LIST,
page=page,
)
if form_list.filtering and form_list.filtering_fields:
add_filter_controls(
keyboard_builder=keyboard_builder,
entity_descriptor=entity_descriptor,
filter=entity_filter,
filtering_fields=form_list.filtering_fields,
)
context = pop_navigation_context(navigation_stack)
if context:
keyboard_builder.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_BACK_BTN)),
callback_data=context.pack(),
)
)
if form_list.caption:
entity_text = get_callable_str(form_list.caption, entity_descriptor)
else:
if entity_descriptor.full_name_plural:
entity_text = get_callable_str(
entity_descriptor.full_name_plural, entity_descriptor
)
else:
entity_text = entity_descriptor.name
if entity_descriptor.description:
entity_text = f"{entity_text} {get_callable_str(entity_descriptor.description, entity_descriptor)}"
state: FSMContext = kwargs["state"]
state_data = kwargs["state_data"]
await state.set_data(state_data)
send_message = get_send_message(message)
await send_message(text=entity_text, reply_markup=keyboard_builder.as_markup())

View File

@@ -0,0 +1,82 @@
from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from babel.support import LazyProxy
from logging import getLogger
from typing import TYPE_CHECKING
from ....model.settings import Settings
from ..context import ContextData, CallbackCommand
from ....utils.main import get_send_message
from ....model.descriptors import EntityCaptionCallable
from ....utils.navigation import save_navigation_context, pop_navigation_context
if TYPE_CHECKING:
from ....main import QBotApp
logger = getLogger(__name__)
router = Router()
@router.callback_query(
ContextData.filter(F.command == CallbackCommand.MENU_ENTRY_ENTITIES)
)
async def menu_entry_entities(message: CallbackQuery, **kwargs):
callback_data: ContextData = kwargs["callback_data"]
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
stack = save_navigation_context(callback_data=callback_data, state_data=state_data)
await entities_menu(message=message, navigation_stack=stack, **kwargs)
async def entities_menu(
message: Message | CallbackQuery,
app: "QBotApp",
state: FSMContext,
navigation_stack: list[ContextData],
**kwargs,
):
keyboard_builder = InlineKeyboardBuilder()
entity_metadata = app.entity_metadata
for entity in entity_metadata.entity_descriptors.values():
if entity.show_in_entities_menu:
if entity.full_name_plural.__class__ == EntityCaptionCallable:
caption = entity.full_name_plural(entity) or entity.name
elif entity.full_name_plural.__class__ == LazyProxy:
caption = f"{f'{entity.icon} ' if entity.icon else ''}{entity.full_name_plural.value or entity.name}"
else:
caption = f"{f'{entity.icon} ' if entity.icon else ''}{entity.full_name_plural or entity.name}"
keyboard_builder.row(
InlineKeyboardButton(
text=caption,
callback_data=ContextData(
command=CallbackCommand.ENTITY_LIST, entity_name=entity.name
).pack(),
)
)
context = pop_navigation_context(navigation_stack)
if context:
keyboard_builder.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_BACK_BTN)),
callback_data=context.pack(),
)
)
state_data = kwargs["state_data"]
await state.set_data(state_data)
send_message = get_send_message(message)
await send_message(
text=(await Settings.get(Settings.APP_STRINGS_REFERENCES)),
reply_markup=keyboard_builder.as_markup(),
)

View File

@@ -0,0 +1,96 @@
from aiogram import Router, F
from aiogram.types import (
Message,
CallbackQuery,
InlineKeyboardButton,
InlineKeyboardMarkup,
)
from aiogram.fsm.context import FSMContext
from aiogram.utils.i18n import I18n
from logging import getLogger
from sqlmodel.ext.asyncio.session import AsyncSession
from ....utils.navigation import pop_navigation_context, save_navigation_context
from ....model.language import LanguageBase
from ....model.settings import Settings
from ....model.user import UserBase
from ..context import ContextData, CallbackCommand
from ..common.routing import route_callback
from ....utils.main import get_send_message
logger = getLogger(__name__)
router = Router()
@router.callback_query(
ContextData.filter(F.command == CallbackCommand.MENU_ENTRY_LANGUAGE)
)
async def menu_entry_language(message: CallbackQuery, **kwargs):
callback_data: ContextData = kwargs["callback_data"]
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
stack = save_navigation_context(callback_data=callback_data, state_data=state_data)
await language_menu(message, navigation_stack=stack, **kwargs)
async def language_menu(
message: Message | CallbackQuery,
navigation_stack: list[ContextData],
user: UserBase,
**kwargs,
):
send_message = get_send_message(message)
inline_keyboard = [
[
InlineKeyboardButton(
text=locale.localized(user.lang.value),
callback_data=ContextData(
command=CallbackCommand.SET_LANGUAGE, data=str(locale)
).pack(),
)
]
for locale in LanguageBase.all_members.values()
]
context = pop_navigation_context(navigation_stack)
if context:
inline_keyboard.append(
[
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_BACK_BTN)),
callback_data=context.pack(),
)
]
)
state: FSMContext = kwargs["state"]
state_data = kwargs["state_data"]
await state.set_data(state_data)
await send_message(
text=(await Settings.get(Settings.APP_STRINGS_LANGUAGE)),
reply_markup=InlineKeyboardMarkup(inline_keyboard=inline_keyboard),
)
@router.callback_query(ContextData.filter(F.command == CallbackCommand.SET_LANGUAGE))
async def set_language(message: CallbackQuery, **kwargs):
user: UserBase = kwargs["user"]
callback_data: ContextData = kwargs["callback_data"]
db_session: AsyncSession = kwargs["db_session"]
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
user.lang = LanguageBase(callback_data.data)
await db_session.commit()
i18n: I18n = kwargs["i18n"]
with i18n.use_locale(user.lang.value):
await route_callback(message, **kwargs)

View File

@@ -0,0 +1,88 @@
from aiogram import Router, F
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from logging import getLogger
from ....model.settings import Settings
from ..context import ContextData, CallbackCommand
from ....utils.main import get_send_message
from ....utils.navigation import save_navigation_context, pop_navigation_context
import qbot.bot.handlers.menu.entities as entities
import qbot.bot.handlers.menu.settings as settings
import qbot.bot.handlers.menu.parameters as parameters
import qbot.bot.handlers.menu.language as language
import qbot.bot.handlers.editors.main as editor
import qbot.bot.handlers.editors.main_callbacks as editor_callbacks
import qbot.bot.handlers.forms.entity_list as entity_list
import qbot.bot.handlers.forms.entity_form as entity_form
import qbot.bot.handlers.forms.entity_form_callbacks as entity_form_callbacks
import qbot.bot.handlers.common.filtering_callbacks as filtering_callbacks
import qbot.bot.handlers.user_handlers.main as user_handlers_main
logger = getLogger(__name__)
router = Router()
@router.callback_query(ContextData.filter(F.command == CallbackCommand.MENU_ENTRY_MAIN))
async def menu_entry_main(message: CallbackQuery, **kwargs):
stack = save_navigation_context(
callback_data=kwargs["callback_data"], state=kwargs["state"]
)
await main_menu(message, navigation_stack=stack, **kwargs)
async def main_menu(
message: Message | CallbackQuery, navigation_stack: list[ContextData], **kwargs
):
keyboard_builder = InlineKeyboardBuilder()
keyboard_builder.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_REFERENCES_BTN)),
callback_data=ContextData(
command=CallbackCommand.MENU_ENTRY_ENTITIES
).pack(),
)
)
keyboard_builder.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_SETTINGS_BTN)),
callback_data=ContextData(
command=CallbackCommand.MENU_ENTRY_SETTINGS
).pack(),
)
)
context = pop_navigation_context(navigation_stack)
if context:
keyboard_builder.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_BACK_BTN)),
callback_data=context.pack(),
)
)
send_message = get_send_message(message)
await send_message(
text=(await Settings.get(Settings.APP_STRINGS_MAIN_NENU)),
reply_markup=keyboard_builder.as_markup(),
)
router.include_routers(
entities.router,
settings.router,
parameters.router,
language.router,
editor.router,
editor_callbacks.router,
entity_list.router,
entity_form.router,
entity_form_callbacks.router,
filtering_callbacks.router,
user_handlers_main.router,
)

View File

@@ -0,0 +1,102 @@
from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from logging import getLogger
from ....model.settings import Settings
from ....model.user import UserBase
from ..context import ContextData, CallbackCommand, CommandContext
from ....utils.main import (
get_send_message,
clear_state,
get_value_repr,
get_callable_str,
)
from ....utils.navigation import save_navigation_context, pop_navigation_context
from ....auth import authorize_command
logger = getLogger(__name__)
router = Router()
@router.callback_query(
ContextData.filter(F.command == CallbackCommand.MENU_ENTRY_PARAMETERS)
)
async def menu_entry_parameters(message: CallbackQuery, **kwargs):
callback_data: ContextData = kwargs["callback_data"]
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
clear_state(state_data=state_data)
stack = save_navigation_context(callback_data=callback_data, state_data=state_data)
await parameters_menu(message=message, navigation_stack=stack, **kwargs)
async def parameters_menu(
message: Message | CallbackQuery,
user: UserBase,
callback_data: ContextData,
navigation_stack: list[ContextData],
**kwargs,
):
if not await authorize_command(user=user, callback_data=callback_data):
await message.answer(await Settings.get(Settings.APP_STRINGS_FORBIDDEN))
settings = await Settings.get_params()
keyboard_builder = InlineKeyboardBuilder()
for key, value in settings.items():
if not key.is_visible:
continue
if key.caption_value:
caption = get_callable_str(
callable_str=key.caption_value, descriptor=key, entity=None, value=value
)
else:
if key.caption:
caption = get_callable_str(
callable_str=key.caption, descriptor=key, entity=None, value=value
)
else:
caption = key.name
if key.type_ is bool:
caption = f"{'【✔︎】' if value else '【 】'} {caption}"
else:
caption = f"{caption}: {get_value_repr(value=value, field_descriptor=key, locale=user.lang)}"
keyboard_builder.row(
InlineKeyboardButton(
text=caption,
callback_data=ContextData(
command=CallbackCommand.FIELD_EDITOR,
context=CommandContext.SETTING_EDIT,
field_name=key.name,
).pack(),
)
)
context = pop_navigation_context(navigation_stack)
if context:
keyboard_builder.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_BACK_BTN)),
callback_data=context.pack(),
)
)
state: FSMContext = kwargs["state"]
state_data = kwargs["state_data"]
await state.set_data(state_data)
send_message = get_send_message(message)
await send_message(
text=(await Settings.get(Settings.APP_STRINGS_PARAMETERS)),
reply_markup=keyboard_builder.as_markup(),
)

View File

@@ -0,0 +1,80 @@
from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from logging import getLogger
from ....model.settings import Settings
from ....model.user import UserBase
from ....utils.main import get_send_message
from ..context import ContextData, CallbackCommand
from ....auth import authorize_command
from ....utils.navigation import save_navigation_context, pop_navigation_context
logger = getLogger(__name__)
router = Router()
@router.callback_query(
ContextData.filter(F.command == CallbackCommand.MENU_ENTRY_SETTINGS)
)
async def menu_entry_settings(message: CallbackQuery, **kwargs):
callback_data: ContextData = kwargs["callback_data"]
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
stack = save_navigation_context(callback_data=callback_data, state_data=state_data)
await settings_menu(message, navigation_stack=stack, **kwargs)
async def settings_menu(
message: Message | CallbackQuery,
user: UserBase,
navigation_stack: list[ContextData],
**kwargs,
):
keyboard_builder = InlineKeyboardBuilder()
if await authorize_command(
user=user,
callback_data=ContextData(command=CallbackCommand.MENU_ENTRY_PARAMETERS),
):
keyboard_builder.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_PARAMETERS_BTN)),
callback_data=ContextData(
command=CallbackCommand.MENU_ENTRY_PARAMETERS
).pack(),
)
)
keyboard_builder.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_LANGUAGE_BTN)),
callback_data=ContextData(
command=CallbackCommand.MENU_ENTRY_LANGUAGE
).pack(),
)
)
context = pop_navigation_context(navigation_stack)
if context:
keyboard_builder.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_BACK_BTN)),
callback_data=context.pack(),
)
)
state: FSMContext = kwargs["state"]
state_data = kwargs["state_data"]
await state.set_data(state_data)
send_message = get_send_message(message)
await send_message(
text=(await Settings.get(Settings.APP_STRINGS_SETTINGS)),
reply_markup=keyboard_builder.as_markup(),
)

View File

@@ -0,0 +1,73 @@
from aiogram import Router
from aiogram.filters import CommandStart
from aiogram.fsm.context import FSMContext
from aiogram.types import Message
from logging import getLogger
from sqlmodel.ext.asyncio.session import AsyncSession
from ...main import QBotApp
from ...model.settings import Settings
from ...model.language import LanguageBase
from ...utils.main import clear_state
logger = getLogger(__name__)
router = Router()
@router.message(CommandStart())
async def start(
message: Message, db_session: AsyncSession, app: QBotApp, state: FSMContext
):
state_data = await state.get_data()
clear_state(state_data=state_data, clear_nav=True)
User = app.user_class
user = await User.get(session=db_session, id=message.from_user.id)
if not user:
msg_text = (await Settings.get(Settings.APP_STRINGS_WELCOME_P_NAME)).format(
name=message.from_user.full_name
)
try:
if message.from_user.language_code in [
item.value for item in LanguageBase.all_members.values()
]:
lang = LanguageBase(message.from_user.language_code)
else:
lang = LanguageBase.EN
user = await User.create(
session=db_session,
obj_in=User(
id=message.from_user.id,
name=message.from_user.full_name,
lang=lang,
is_active=True,
),
commit=True,
)
except Exception as e:
logger.error("Error creating user", exc_info=True)
message.answer(
(
await Settings.get(Settings.APP_STRINGS_INTERNAL_ERROR_P_ERROR)
).format(error=str(e))
)
return
else:
if user.is_active:
msg_text = (
await Settings.get(Settings.APP_STRINGS_GREETING_P_NAME)
).format(name=user.name)
else:
msg_text = (
await Settings.get(Settings.APP_STRINGS_USER_BLOCKED_P_NAME)
).format(name=user.name)
await message.answer(msg_text)

View File

@@ -0,0 +1,159 @@
from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton
from inspect import iscoroutinefunction
from typing import TYPE_CHECKING
from qbot.utils.main import clear_state
from qbot.utils.navigation import (
save_navigation_context,
get_navigation_context,
pop_navigation_context,
)
from qbot.bot.handlers.editors.main import field_editor
from qbot.bot.handlers.common.routing import route_callback
from qbot.utils.serialization import deserialize
from qbot.utils.main import get_send_message
from qbot.model.descriptors import BotCommand, CommandCallbackContext
from qbot.model.settings import Settings
if TYPE_CHECKING:
from qbot.main import QBotApp
from ..context import ContextData, CallbackCommand, CommandContext
router = Router()
@router.message(F.text.startswith("/"))
async def command_text(message: Message, **kwargs):
str_command = message.text.lstrip("/")
callback_data = ContextData(
command=CallbackCommand.USER_COMMAND, user_command=str_command
)
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
await process_command_handler(
message=message, callback_data=callback_data, **kwargs
)
@router.callback_query(ContextData.filter(F.command == CallbackCommand.USER_COMMAND))
async def command_callback(message: CallbackQuery, **kwargs):
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
await process_command_handler(message=message, **kwargs)
async def process_command_handler(message: Message | CallbackQuery, **kwargs):
state_data: dict = kwargs["state_data"]
callback_data: ContextData = kwargs["callback_data"]
app: "QBotApp" = kwargs["app"]
cmd = app.bot_commands.get(callback_data.user_command.split("&")[0])
if cmd is None:
return
if cmd.clear_navigation:
state_data.pop("navigation_stack", None)
state_data.pop("navigation_context", None)
if cmd.register_navigation:
clear_state(state_data=state_data)
save_navigation_context(callback_data=callback_data, state_data=state_data)
await cammand_handler(message=message, cmd=cmd, **kwargs)
async def cammand_handler(message: Message | CallbackQuery, cmd: BotCommand, **kwargs):
callback_data: ContextData = kwargs["callback_data"]
state: FSMContext = kwargs["state"]
state_data: dict = kwargs["state_data"]
app: "QBotApp" = kwargs["app"]
entity_data_dict: dict = state_data.get("entity_data")
form_data = (
{
key: await deserialize(
session=kwargs["db_session"],
type_=cmd.param_form[key].type_,
value=value,
)
for key, value in entity_data_dict.items()
}
if entity_data_dict and cmd.param_form
else None
)
callback_context = CommandCallbackContext(
message=message,
callback_data=callback_data,
form_data=form_data,
db_session=kwargs["db_session"],
user=kwargs["user"],
app=app,
state_data=state_data,
state=state,
i18n=kwargs["i18n"],
register_navigation=cmd.register_navigation,
kwargs=kwargs,
)
if cmd.pre_check and (not cmd.param_form or (cmd.param_form and form_data is None)):
if iscoroutinefunction(cmd.pre_check):
if not await cmd.pre_check(callback_context):
return
else:
if not cmd.pre_check(callback_context):
return
if form_data is None and cmd.param_form:
field_descriptor = list(cmd.param_form.values())[0]
kwargs["callback_data"] = ContextData(
command=CallbackCommand.FIELD_EDITOR,
context=CommandContext.COMMAND_FORM,
field_name=field_descriptor.name,
user_command=callback_data.user_command,
)
return await field_editor(message=message, **kwargs)
await cmd.handler(callback_context)
if callback_context.register_navigation:
await state.set_data(state_data)
stack, navigation_context = get_navigation_context(state_data=state_data)
back_callback_data = pop_navigation_context(stack=stack)
if back_callback_data:
callback_context.keyboard_builder.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_BACK_BTN)),
callback_data=back_callback_data.pack(),
)
)
send_message = get_send_message(message)
if isinstance(message, CallbackQuery):
message = message.message
if callback_context.message_text:
await send_message(
text=callback_context.message_text,
reply_markup=callback_context.keyboard_builder.as_markup(),
)
else:
await message.edit_reply_markup(
reply_markup=callback_context.keyboard_builder.as_markup()
)
else:
clear_state(state_data=state_data)
await route_callback(message, back=True, **kwargs)

View File

@@ -0,0 +1,79 @@
from pydantic import computed_field, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import Literal, Self
import warnings
class Config(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env", env_ignore_empty=True, extra="ignore"
)
SECRET_KEY: str = "changethis"
ENVIRONMENT: Literal["local", "staging", "production"] = "local"
DB_NAME: str = "app"
DB_HOST: str = "db"
DB_PORT: int = 5432
DB_USER: str = "app"
DB_PASSWORD: str = "changethis"
@computed_field
@property
def DATABASE_URI(self) -> str:
return f"postgresql+asyncpg://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
DOMAIN: str = "localhost"
@computed_field
@property
def API_DOMAIN(self) -> str:
if self.ENVIRONMENT == "local":
return self.DOMAIN
return f"api.{self.DOMAIN}"
@computed_field
@property
def API_URL(self) -> str:
if self.USE_NGROK:
return self.NGROK_URL
return (
f"{'http' if self.ENVIRONMENT == 'local' else 'https'}://{self.API_DOMAIN}"
)
API_PORT: int = 8000
TELEGRAM_BOT_TOKEN: str = "changethis"
ADMIN_TELEGRAM_ID: int
USE_NGROK: bool = False
NGROK_AUTH_TOKEN: str = "changethis"
NGROK_URL: str = ""
LOG_LEVEL: str = "DEBUG"
def _check_default_secret(self, var_name: str, value: str | None) -> None:
if value == "changethis":
message = (
f'The value of {var_name} is "changethis", '
"for security, please change it, at least for deployments."
)
if self.ENVIRONMENT == "local":
warnings.warn(message, stacklevel=1)
else:
raise ValueError(message)
@model_validator(mode="after")
def _enforce_non_default_secrets(self) -> Self:
self._check_default_secret("SECRET_KEY", self.SECRET_KEY)
self._check_default_secret("DB_PASSWORD", self.DB_PASSWORD)
self._check_default_secret("TELEGRAM_BOT_TOKEN", self.TELEGRAM_BOT_TOKEN)
if self.USE_NGROK:
self._check_default_secret("NGROK_AUTH_TOKEN", self.NGROK_AUTH_TOKEN)
return self
config = Config()

19
src/qbot/db/__init__.py Normal file
View File

@@ -0,0 +1,19 @@
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
from ..config import config
# import logging
# logger = logging.getLogger('sqlalchemy.engine')
# logger.setLevel(logging.DEBUG)
async_engine = create_async_engine(config.DATABASE_URI)
async_session = sessionmaker[AsyncSession](
async_engine, class_=AsyncSession, expire_on_commit=False
)
async def get_db() -> AsyncSession: # type: ignore
async with async_session() as session:
yield session

View File

@@ -0,0 +1,84 @@
from aiogram.fsm.state import State
from aiogram.fsm.storage.base import (
BaseStorage,
StorageKey,
StateType,
DefaultKeyBuilder,
KeyBuilder,
)
from sqlmodel import select
from typing import Any, Dict
import ujson as json
from ..db import async_session
from ..model.fsm_storage import FSMStorage
class DbStorage(BaseStorage):
def __init__(self, key_builder: KeyBuilder | None = None) -> None:
if key_builder is None:
key_builder = DefaultKeyBuilder()
self.key_builder = key_builder
async def set_state(self, key: StorageKey, state: StateType = None) -> None:
db_key = self.key_builder.build(key, "state")
async with async_session() as session:
db_state = (
await session.exec(select(FSMStorage).where(FSMStorage.key == db_key))
).first()
if db_state:
if state is None:
await session.delete(db_state)
else:
db_state.value = state.state if isinstance(state, State) else state
elif state is not None:
db_state = FSMStorage(
key=db_key, value=state.state if isinstance(state, State) else state
)
session.add(db_state)
else:
return
await session.commit()
async def get_state(self, key: StorageKey) -> str | None:
db_key = self.key_builder.build(key, "state")
async with async_session() as session:
db_state = (
await session.exec(select(FSMStorage).where(FSMStorage.key == db_key))
).first()
return db_state.value if db_state else None
async def set_data(self, key: StorageKey, data: Dict[str, Any]) -> None:
db_key = self.key_builder.build(key, "data")
async with async_session() as session:
db_data = (
await session.exec(select(FSMStorage).where(FSMStorage.key == db_key))
).first()
if db_data:
if not data:
await session.delete(db_data)
else:
db_data.value = json.dumps(data, ensure_ascii=False)
elif data:
db_data = FSMStorage(
key=db_key, value=json.dumps(data, ensure_ascii=False)
)
session.add(db_data)
else:
return
await session.commit()
async def get_data(self, key: StorageKey) -> Dict[str, Any]:
db_key = self.key_builder.build(key, "data")
async with async_session() as session:
db_data = (
await session.exec(select(FSMStorage).where(FSMStorage.key == db_key))
).first()
return json.loads(db_data.value) if db_data else {}
async def close(self):
return await super().close()

View File

@@ -0,0 +1,23 @@
from aiogram.types import InlineKeyboardButton
from ..utils.navigation import pop_navigation_context
from ..model.descriptors import CommandCallbackContext
from ..model.settings import Settings
async def get_back_button(
context: CommandCallbackContext, text: str = None
) -> InlineKeyboardButton | None:
stack = context.state_data.get("navigation_stack")
if not stack:
return None
back_callback_data = pop_navigation_context(stack)
if not text:
text = await Settings.get(Settings.APP_STRINGS_BACK_BTN)
return InlineKeyboardButton(
text=text,
callback_data=back_callback_data.pack(),
)

209
src/qbot/main.py Normal file
View File

@@ -0,0 +1,209 @@
from contextlib import asynccontextmanager
from typing import Annotated, Callable, Any
from typing_extensions import Doc
from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.types import Message, BotCommand as AiogramBotCommand
from aiogram.utils.callback_answer import CallbackAnswerMiddleware
from aiogram.utils.i18n import I18n
from fastapi import FastAPI
from fastapi.applications import Lifespan, AppType
from secrets import token_hex
from logging import getLogger
from .config import Config
from .fsm.db_storage import DbStorage
from .middleware.telegram import AuthMiddleware, I18nMiddleware
from .model.user import UserBase
from .model.descriptors import BotCommand
from .router import Router
logger = getLogger(__name__)
@asynccontextmanager
async def default_lifespan(app: "QBotApp"):
logger.debug("starting qbot app")
if app.lifespan_bot_init:
if app.config.USE_NGROK:
app.ngrok_init()
await app.bot_init()
logger.info("qbot app started")
if app.lifespan:
async with app.lifespan(app):
yield
else:
yield
logger.info("stopping qbot app")
if app.lifespan_bot_init:
await app.bot_close()
if app.config.USE_NGROK:
app.ngrok_stop()
logger.info("qbot app stopped")
class QBotApp[UserType: UserBase](FastAPI):
"""
`QBotApp` app class, the main entrypoint to use QBot.
Derives from FastAPI.
## Example
```python
from qbot import QbotApp
app = QBotApp()
```
"""
def __init__(
self,
user_class: Annotated[
type[UserType],
Doc(
"""
User entity class, derived from :class:`UserBase`.
If not provided, default :class:`DefaultUser` will be used.
"""
),
] = None,
config: Config | None = None,
bot_start: Callable[
[
Callable[[Message, Any], tuple[UserType, bool]],
Message,
Any,
],
None,
] = None,
lifespan: Lifespan[AppType] | None = None,
lifespan_bot_init: bool = True,
allowed_updates: list[str] | None = None,
*args,
**kwargs,
):
if config is None:
config = Config()
self.allowed_updates = allowed_updates or ["message", "callback_query"]
self.user_class = user_class
self.entity_metadata = user_class.entity_metadata
self.config = config
self.lifespan = lifespan
self.bot = Bot(
token=self.config.TELEGRAM_BOT_TOKEN,
default=DefaultBotProperties(parse_mode="HTML"),
)
dp = Dispatcher(storage=DbStorage())
i18n = I18n(path="locales", default_locale="en", domain="messages")
i18n_middleware = I18nMiddleware(user_class=user_class, i18n=i18n)
i18n_middleware.setup(dp)
dp.callback_query.middleware(CallbackAnswerMiddleware())
from .bot.handlers.start import router as start_router
dp.include_router(start_router)
from .bot.handlers.menu.main import router as main_menu_router
auth = AuthMiddleware(user_class=user_class)
main_menu_router.message.middleware.register(auth)
main_menu_router.callback_query.middleware.register(auth)
dp.include_router(main_menu_router)
self.dp = dp
self.bot_auth_token = token_hex(128)
self.start_handler = bot_start
self.bot_commands = dict[str, BotCommand]()
self.lifespan_bot_init = lifespan_bot_init
super().__init__(lifespan=default_lifespan, *args, **kwargs)
from .api_route.telegram import router as telegram_router
self.include_router(telegram_router, prefix="/api/telegram", tags=["telegram"])
self.root_router = Router()
self.root_router._commands = self.bot_commands
self.command = self.root_router.command
def register_routers(self, *routers: Router):
for router in routers:
for command_name, command in router._commands.items():
self.bot_commands[command_name] = command
def ngrok_init(self):
try:
from pyngrok import ngrok
from pyngrok.conf import PyngrokConfig
except ImportError:
logger.error("pyngrok is not installed")
raise
tunnel = ngrok.connect(
self.config.API_PORT,
pyngrok_config=PyngrokConfig(auth_token=self.config.NGROK_AUTH_TOKEN),
)
self.config.NGROK_URL = tunnel.public_url
def ngrok_stop(self):
try:
from pyngrok import ngrok
except ImportError:
logger.error("pyngrok is not installed")
raise
ngrok.disconnect(self.config.NGROK_URL)
ngrok.kill()
async def bot_init(self):
commands_captions = dict[str, list[tuple[str, str]]]()
for command_name, command in self.bot_commands.items():
if command.show_in_bot_commands:
if isinstance(command.caption, str) or command.caption is None:
if "default" not in commands_captions:
commands_captions["default"] = []
commands_captions["default"].append(
(command_name, command.caption or command_name)
)
else:
for locale, description in command.caption.items():
if locale not in commands_captions:
commands_captions[locale] = []
commands_captions[locale].append((command_name, description))
for locale, commands in commands_captions.items():
await self.bot.set_my_commands(
[
AiogramBotCommand(command=command[0], description=command[1])
for command in commands
],
language_code=None if locale == "default" else locale,
)
await self.bot.set_webhook(
url=f"{self.config.API_URL}/api/telegram/webhook",
drop_pending_updates=True,
allowed_updates=self.allowed_updates,
secret_token=self.bot_auth_token,
)
async def bot_close(self):
await self.bot.delete_webhook()

View File

@@ -0,0 +1,2 @@
from .auth import AuthMiddleware as AuthMiddleware
from .i18n import I18nMiddleware as I18nMiddleware

View File

@@ -0,0 +1,37 @@
from aiogram import BaseMiddleware
from aiogram.types import TelegramObject, Message, CallbackQuery
from babel.support import LazyProxy
from typing import Any, Awaitable, Callable, Dict
from ...model.user import UserBase
class AuthMiddleware(BaseMiddleware):
def __init__[UserType: UserBase](
self,
user_class: type[UserType],
not_authenticated_msg: LazyProxy | str = "not authenticated",
not_active_msg: LazyProxy | str = "not active",
):
self.user_class = user_class
self.not_authenticated_msg = not_authenticated_msg
self.not_active_msg = not_active_msg
async def __call__(
self,
handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
event: TelegramObject,
data: Dict[str, Any],
) -> Any:
user = await self.user_class.get(
id=event.from_user.id, session=data["db_session"]
)
if user and user.is_active:
data["user"] = user
return await handler(event, data)
if type(event) in [Message, CallbackQuery]:
if user and not user.is_active:
return await event.answer(self.not_active_msg)
return await event.answer(self.not_authenticated_msg)

View File

@@ -0,0 +1,24 @@
from typing import Optional, Dict, Any
from aiogram.utils.i18n import I18n, SimpleI18nMiddleware
from aiogram.types import TelegramObject
from ...model.user import UserBase
class I18nMiddleware(SimpleI18nMiddleware):
def __init__[UserType: UserBase](
self,
user_class: type[UserType],
i18n: I18n,
i18n_key: Optional[str] = "i18n",
middleware_key: str = "i18n_middleware",
) -> None:
self.user_class = user_class
super().__init__(i18n=i18n, i18n_key=i18n_key, middleware_key=middleware_key)
async def get_locale(self, event: TelegramObject, data: Dict[str, Any]) -> str:
db_session = data.get("db_session")
if db_session and event.model_fields.get("from_user"):
user = await self.user_class.get(id=event.from_user.id, session=db_session)
if user and user.lang:
return user.lang.value
return await super().get_locale(event=event, data=data)

View File

@@ -0,0 +1,26 @@
# from typing import Any, Awaitable, Callable, Dict
# from aiogram import BaseMiddleware
# from aiogram.types import TelegramObject
# from aiogram.fsm.context import FSMContext
# from aiogram.utils.i18n import gettext as _
# from ...bot.handlers.context import ContextData
# class ResetStateMiddleware(BaseMiddleware):
# async def __call__(self,
# handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
# event: TelegramObject,
# data: Dict[str, Any]) -> Any:
# save_state = False
# callback_data = data.get("callback_data")
# if isinstance(callback_data, ContextData):
# save_state = callback_data.save_state
# if not save_state:
# state = data.get("state")
# if isinstance(state, FSMContext):
# await state.clear()
# return await handler(event, data)

View File

@@ -0,0 +1,43 @@
from functools import wraps
from sqlalchemy.inspection import inspect
from sqlalchemy.orm.state import InstanceState
from typing import cast
from .bot_enum import BotEnum, EnumMember
from ..db import async_session
class EntityPermission(BotEnum):
LIST = EnumMember("list")
READ = EnumMember("read")
CREATE = EnumMember("create")
UPDATE = EnumMember("update")
DELETE = EnumMember("delete")
LIST_ALL = EnumMember("list_all")
READ_ALL = EnumMember("read_all")
CREATE_ALL = EnumMember("create_all")
UPDATE_ALL = EnumMember("update_all")
DELETE_ALL = EnumMember("delete_all")
def session_dep(func):
@wraps(func)
async def wrapper(cls, *args, **kwargs):
if "session" in kwargs and kwargs["session"]:
return await func(cls, *args, **kwargs)
_session = None
state = cast(InstanceState, inspect(cls))
if hasattr(state, "async_session"):
_session = state.async_session
if not _session:
async with async_session() as session:
kwargs["session"] = session
return await func(cls, *args, **kwargs)
else:
kwargs["session"] = _session
return await func(cls, *args, **kwargs)
return wrapper

View File

@@ -0,0 +1,7 @@
class Singleton(type):
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
return cls._instances[cls]

View File

@@ -0,0 +1,432 @@
from types import NoneType, UnionType
from typing import (
Any,
ClassVar,
ForwardRef,
Optional,
Self,
Union,
get_args,
get_origin,
TYPE_CHECKING,
dataclass_transform,
)
from pydantic import BaseModel
from sqlmodel import SQLModel, BigInteger, Field, select, func, column, col
from sqlmodel.main import FieldInfo
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel.sql.expression import SelectOfScalar
from sqlmodel.main import SQLModelMetaclass, RelationshipInfo
from .descriptors import EntityDescriptor, EntityField, FieldDescriptor, Filter
from .entity_metadata import EntityMetadata
from . import session_dep
if TYPE_CHECKING:
from .user import UserBase
@dataclass_transform(
kw_only_default=True,
field_specifiers=(Field, FieldInfo, EntityField, FieldDescriptor),
)
class BotEntityMetaclass(SQLModelMetaclass):
_future_references = {}
def __new__(mcs, name, bases, namespace, **kwargs):
bot_fields_descriptors = {}
if bases:
bot_entity_descriptor = bases[0].__dict__.get("bot_entity_descriptor")
bot_fields_descriptors = (
{
key: FieldDescriptor(**value.__dict__.copy())
for key, value in bot_entity_descriptor.fields_descriptors.items()
}
if bot_entity_descriptor
else {}
)
if "__annotations__" in namespace:
for annotation in namespace["__annotations__"]:
if annotation in ["bot_entity_descriptor", "entity_metadata"]:
continue
attribute_value = namespace.get(annotation)
if isinstance(attribute_value, RelationshipInfo):
continue
descriptor_kwargs = {}
descriptor_name = annotation
if attribute_value:
if isinstance(attribute_value, EntityField):
descriptor_kwargs = attribute_value.__dict__.copy()
sm_descriptor = descriptor_kwargs.pop("sm_descriptor", None)
if sm_descriptor:
namespace[annotation] = sm_descriptor
else:
namespace.pop(annotation)
descriptor_name = descriptor_kwargs.pop("name") or annotation
type_ = namespace["__annotations__"][annotation]
type_origin = get_origin(type_)
field_descriptor = FieldDescriptor(
name=descriptor_name,
field_name=annotation,
type_=type_,
type_base=type_,
**descriptor_kwargs,
)
is_list = False
is_optional = False
if type_origin is list:
field_descriptor.is_list = is_list = True
field_descriptor.type_base = type_ = get_args(type_)[0]
if type_origin is Union:
args = get_args(type_)
if isinstance(args[0], ForwardRef):
field_descriptor.is_optional = is_optional = True
field_descriptor.type_base = type_ = args[0].__forward_arg__
elif args[1] is NoneType:
field_descriptor.is_optional = is_optional = True
field_descriptor.type_base = type_ = args[0]
if type_origin is UnionType and get_args(type_)[1] is NoneType:
field_descriptor.is_optional = is_optional = True
field_descriptor.type_base = type_ = get_args(type_)[0]
if isinstance(type_, str):
type_not_found = True
for (
entity_descriptor
) in EntityMetadata().entity_descriptors.values():
if type_ == entity_descriptor.class_name:
field_descriptor.type_base = entity_descriptor.type_
field_descriptor.type_ = (
list[entity_descriptor.type_]
if is_list
else (
Optional[entity_descriptor.type_]
if type_origin == Union and is_optional
else (
entity_descriptor.type_ | None
if (type_origin == UnionType and is_optional)
else entity_descriptor.type_
)
)
)
type_not_found = False
break
if type_not_found:
if type_ in mcs._future_references:
mcs._future_references[type_].append(field_descriptor)
else:
mcs._future_references[type_] = [field_descriptor]
bot_fields_descriptors[descriptor_name] = field_descriptor
descriptor_name = name
if "bot_entity_descriptor" in namespace:
entity_descriptor = namespace.pop("bot_entity_descriptor")
descriptor_kwargs: dict = entity_descriptor.__dict__.copy()
descriptor_name = descriptor_kwargs.pop("name", None)
descriptor_name = descriptor_name or name.lower()
namespace["bot_entity_descriptor"] = EntityDescriptor(
name=descriptor_name,
class_name=name,
type_=name,
fields_descriptors=bot_fields_descriptors,
**descriptor_kwargs,
)
else:
descriptor_name = name.lower()
namespace["bot_entity_descriptor"] = EntityDescriptor(
name=descriptor_name,
class_name=name,
type_=name,
fields_descriptors=bot_fields_descriptors,
)
descriptor_fields_sequence = [
key
for key, val in bot_fields_descriptors.items()
if not (val.is_optional or val.name == "id")
]
entity_descriptor: EntityDescriptor = namespace["bot_entity_descriptor"]
if entity_descriptor.default_form.edit_field_sequence is None:
entity_descriptor.default_form.edit_field_sequence = (
descriptor_fields_sequence
)
for form in entity_descriptor.forms.values():
if form.edit_field_sequence is None:
form.edit_field_sequence = descriptor_fields_sequence
for field_descriptor in bot_fields_descriptors.values():
field_descriptor.entity_descriptor = namespace["bot_entity_descriptor"]
if "table" not in kwargs:
kwargs["table"] = True
if kwargs["table"]:
entity_metadata = EntityMetadata()
entity_metadata.entity_descriptors[descriptor_name] = namespace[
"bot_entity_descriptor"
]
if "__annotations__" in namespace:
namespace["__annotations__"]["entity_metadata"] = ClassVar[
EntityMetadata
]
else:
namespace["__annotations__"] = {
"entity_metadata": ClassVar[EntityMetadata]
}
namespace["entity_metadata"] = entity_metadata
type_ = super().__new__(mcs, name, bases, namespace, **kwargs)
if name in mcs._future_references:
for field_descriptor in mcs._future_references[name]:
type_origin = get_origin(field_descriptor.type_)
field_descriptor.type_base = type_
field_descriptor.type_ = (
list[type_]
if type_origin is list
else (
Optional[type_]
if type_origin == Union
and isinstance(get_args(field_descriptor.type_)[0], ForwardRef)
else type_ | None
if type_origin == UnionType
else type_
)
)
setattr(namespace["bot_entity_descriptor"], "type_", type_)
return type_
class BotEntity[CreateSchemaType: BaseModel, UpdateSchemaType: BaseModel](
SQLModel, metaclass=BotEntityMetaclass, table=False
):
bot_entity_descriptor: ClassVar[EntityDescriptor]
entity_metadata: ClassVar[EntityMetadata]
id: int = EntityField(
sm_descriptor=Field(primary_key=True, sa_type=BigInteger), is_visible=False
)
@classmethod
@session_dep
async def get(cls, *, session: AsyncSession | None = None, id: int):
return await session.get(cls, id, populate_existing=True)
@classmethod
def _static_filter_condition(
cls, select_statement: SelectOfScalar[Self], static_filter: list[Filter]
):
for sfilt in static_filter:
column = getattr(cls, sfilt.field_name)
if sfilt.operator == "==":
condition = column.__eq__(sfilt.value)
elif sfilt.operator == "!=":
condition = column.__ne__(sfilt.value)
elif sfilt.operator == "<":
condition = column.__lt__(sfilt.value)
elif sfilt.operator == "<=":
condition = column.__le__(sfilt.value)
elif sfilt.operator == ">":
condition = column.__gt__(sfilt.value)
elif sfilt.operator == ">=":
condition = column.__ge__(sfilt.value)
elif sfilt.operator == "ilike":
condition = col(column).ilike(f"%{sfilt.value}%")
elif sfilt.operator == "like":
condition = col(column).like(f"%{sfilt.value}%")
elif sfilt.operator == "in":
condition = col(column).in_(sfilt.value)
elif sfilt.operator == "not in":
condition = col(column).notin_(sfilt.value)
elif sfilt.operator == "is none":
condition = col(column).is_(None)
elif sfilt.operator == "is not none":
condition = col(column).isnot(None)
elif sfilt.operator == "contains":
condition = sfilt.value == col(column).any_()
else:
condition = None
if condition is not None:
select_statement = select_statement.where(condition)
return select_statement
@classmethod
def _filter_condition(
cls,
select_statement: SelectOfScalar[Self],
filter: str,
filter_fields: list[str],
):
condition = None
for field in filter_fields:
if condition is not None:
condition = condition | (column(field).ilike(f"%{filter}%"))
else:
condition = column(field).ilike(f"%{filter}%")
return select_statement.where(condition)
@classmethod
@session_dep
async def get_count(
cls,
*,
session: AsyncSession | None = None,
static_filter: list[Filter] | Any = None,
filter: str = None,
filter_fields: list[str] = None,
ext_filter: Any = None,
user: "UserBase" = None,
) -> int:
select_statement = select(func.count()).select_from(cls)
if static_filter:
if isinstance(static_filter, list):
select_statement = cls._static_filter_condition(
select_statement, static_filter
)
else:
select_statement = select_statement.where(static_filter)
if filter and filter_fields:
select_statement = cls._filter_condition(
select_statement, filter, filter_fields
)
if ext_filter:
select_statement = select_statement.where(ext_filter)
if user:
select_statement = cls._ownership_condition(select_statement, user)
return await session.scalar(select_statement)
@classmethod
@session_dep
async def get_multi(
cls,
*,
session: AsyncSession | None = None,
order_by=None,
static_filter: list[Filter] | Any = None,
filter: str = None,
filter_fields: list[str] = None,
ext_filter: Any = None,
user: "UserBase" = None,
skip: int = 0,
limit: int = None,
):
select_statement = select(cls).offset(skip)
if limit:
select_statement = select_statement.limit(limit)
if static_filter is not None:
if isinstance(static_filter, list):
select_statement = cls._static_filter_condition(
select_statement, static_filter
)
else:
select_statement = select_statement.where(static_filter)
if filter and filter_fields:
select_statement = cls._filter_condition(
select_statement, filter, filter_fields
)
if ext_filter is not None:
select_statement = select_statement.where(ext_filter)
if user:
select_statement = cls._ownership_condition(select_statement, user)
if order_by is not None:
select_statement = select_statement.order_by(order_by)
return (await session.exec(select_statement)).all()
@classmethod
def _ownership_condition(
cls, select_statement: SelectOfScalar[Self], user: "UserBase"
):
if cls.bot_entity_descriptor.ownership_fields:
condition = None
for role in user.roles:
if role in cls.bot_entity_descriptor.ownership_fields:
owner_col = column(cls.bot_entity_descriptor.ownership_fields[role])
if condition is not None:
condition = condition | (owner_col == user.id)
else:
condition = owner_col == user.id
else:
condition = None
break
if condition is not None:
return select_statement.where(condition)
return select_statement
@classmethod
@session_dep
async def create(
cls,
*,
session: AsyncSession | None = None,
obj_in: CreateSchemaType,
commit: bool = False,
):
if isinstance(obj_in, cls):
obj = obj_in
else:
obj = cls(**obj_in.model_dump())
session.add(obj)
if commit:
await session.commit()
return obj
@classmethod
@session_dep
async def update(
cls,
*,
session: AsyncSession | None = None,
id: int,
obj_in: UpdateSchemaType,
commit: bool = False,
):
obj = await session.get(cls, id)
if obj:
obj_data = obj.model_dump()
update_data = obj_in.model_dump(exclude_unset=True)
for field in obj_data:
if field in update_data:
setattr(obj, field, update_data[field])
session.add(obj)
if commit:
await session.commit()
return obj
return None
@classmethod
@session_dep
async def remove(
cls, *, session: AsyncSession | None = None, id: int, commit: bool = False
):
obj = await session.get(cls, id)
if obj:
await session.delete(obj)
if commit:
await session.commit()
return obj
return None

181
src/qbot/model/bot_enum.py Normal file
View File

@@ -0,0 +1,181 @@
from aiogram.utils.i18n import I18n
from pydantic_core.core_schema import str_schema
from sqlalchemy.types import TypeDecorator
from sqlmodel import AutoString
from typing import Any, Self, overload
class BotEnumMetaclass(type):
def __new__(cls, name: str, bases: tuple[type], namespace: dict[str, Any]):
all_members = {}
if (
bases
and bases[0].__name__ != "BotEnum"
and "all_members" in bases[0].__dict__
):
all_members = bases[0].__dict__["all_members"]
annotations = {}
for key, value in namespace.items():
if key.isupper() and not key.startswith("__") and not key.endswith("__"):
if not isinstance(value, EnumMember):
value = EnumMember(value, None)
if key in all_members.keys() and all_members[key].value != value.value:
raise ValueError(
f"Enum member {key} already exists with different value. Use same value to extend it."
)
if (
value.value in [member.value for member in all_members.values()]
and key not in all_members.keys()
):
raise ValueError(f"Duplicate enum value {value[0]}")
member = EnumMember(
value=value.value,
loc_obj=value.loc_obj,
parent=None,
name=key,
casting=False,
)
namespace[key] = member
all_members[key] = member
annotations[key] = type(member)
namespace["__annotations__"] = annotations
namespace["all_members"] = all_members
type_ = super().__new__(cls, name, bases, namespace)
for key, value in all_members.items():
if not value._parent:
value._parent = type_
return type_
class EnumMember(object):
@overload
def __init__(self, value: str) -> "EnumMember": ...
@overload
def __init__(self, value: "EnumMember") -> "EnumMember": ...
@overload
def __init__(self, value: str, loc_obj: dict[str, str]) -> "EnumMember": ...
def __init__(
self,
value: str = None,
loc_obj: dict[str, str] = None,
parent: type = None,
name: str = None,
casting: bool = True,
) -> "EnumMember":
if not casting:
self._parent = parent
self._name = name
self.value = value
self.loc_obj = loc_obj
@overload
def __new__(cls: Self, *args, **kwargs) -> "EnumMember": ...
def __new__(cls, *args, casting: bool = True, **kwargs) -> "EnumMember":
if (cls.__name__ == "EnumMember") or not casting:
obj = super().__new__(cls)
kwargs["casting"] = False
obj.__init__(*args, **kwargs)
return obj
if args.__len__() == 0:
return list(cls.all_members.values())[0]
if args.__len__() == 1 and isinstance(args[0], str):
return {member.value: member for key, member in cls.all_members.items()}[
args[0]
]
elif args.__len__() == 1:
return {member.value: member for key, member in cls.all_members.items()}[
args[0].value
]
else:
return args[0]
def __get_pydantic_core_schema__(cls, *args, **kwargs):
return str_schema()
def __get__(self, instance, owner) -> Self:
return {
member.value: member for key, member in self._parent.all_members.items()
}[self.value]
def __set__(self, instance, value):
instance.__dict__[self] = value
def __repr__(self):
return f"<{self._parent.__name__ if self._parent else 'EnumMember'}.{self._name}: '{self.value}'>"
def __str__(self):
return self.value
def __eq__(self, other: Self | str | Any | None) -> bool:
if other is None:
return False
if isinstance(other, str):
return self.value == other
if isinstance(other, EnumMember):
return self.value == other.value and (
issubclass(self._parent, other._parent)
or issubclass(other._parent, self._parent)
)
return other.__eq__(self.value)
def __hash__(self):
return hash(self.value)
def localized(self, lang: str = None) -> str:
if self.loc_obj:
if not lang:
i18n = I18n.get_current()
if i18n:
lang = i18n.current_locale
else:
lang = list(self.loc_obj.keys())[0]
if lang in self.loc_obj.keys():
return self.loc_obj[lang]
else:
return self.loc_obj[list(self.loc_obj.keys())[0]]
return self.value
class BotEnum(EnumMember, metaclass=BotEnumMetaclass):
all_members: dict[str, EnumMember]
class EnumType(TypeDecorator):
impl = AutoString
cache_ok = True
def __init__(self, enum_type: BotEnum):
self._enum_type = enum_type
super().__init__()
def _process_param(self, value):
if value and isinstance(value, EnumMember):
return value.value
return str(value)
def process_bind_param(self, value, dialect):
return self._process_param(value)
def process_result_value(self, value, dialect):
if value:
return self._enum_type(value)
return None
def process_literal_param(self, value, dialect):
return self._process_param(value)

View File

@@ -0,0 +1,4 @@
from .user import UserBase
class DefaultUser(UserBase): ...

View File

@@ -0,0 +1,212 @@
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton
from aiogram.fsm.context import FSMContext
from aiogram.utils.i18n import I18n
from aiogram.utils.keyboard import InlineKeyboardBuilder
from typing import Any, Callable, TYPE_CHECKING, Literal, Union
from babel.support import LazyProxy
from dataclasses import dataclass, field
from sqlmodel.ext.asyncio.session import AsyncSession
from .role import RoleBase
from . import EntityPermission
from ..bot.handlers.context import ContextData
if TYPE_CHECKING:
from .bot_entity import BotEntity
from ..main import QBotApp
from .user import UserBase
EntityCaptionCallable = Callable[["EntityDescriptor"], str]
EntityItemCaptionCallable = Callable[["EntityDescriptor", Any], str]
EntityFieldCaptionCallable = Callable[["FieldDescriptor", Any, Any], str]
@dataclass
class FieldEditButton:
field_name: str
caption: str | LazyProxy | EntityFieldCaptionCallable | None = None
visibility: Callable[[Any], bool] | None = None
@dataclass
class CommandButton:
command: ContextData | Callable[[ContextData, Any], ContextData] | str
caption: str | LazyProxy | EntityItemCaptionCallable
visibility: Callable[[Any], bool] | None = None
@dataclass
class InlineButton:
inline_button: InlineKeyboardButton | Callable[[Any], InlineKeyboardButton]
visibility: Callable[[Any], bool] | None = None
@dataclass
class Filter:
field_name: str
operator: Literal[
"==",
"!=",
">",
"<",
">=",
"<=",
"in",
"not in",
"like",
"ilike",
"is none",
"is not none",
"contains",
]
value_type: Literal["const", "param"] = "const"
value: Any | None = None
param_index: int | None = None
@dataclass
class EntityList:
caption: str | LazyProxy | EntityCaptionCallable | None = None
item_repr: EntityItemCaptionCallable | None = None
show_add_new_button: bool = True
item_form: str | None = None
pagination: bool = True
static_filters: list[Filter] = None
filtering: bool = False
filtering_fields: list[str] = None
order_by: str | Any | None = None
@dataclass
class EntityForm:
item_repr: EntityItemCaptionCallable | None = None
edit_field_sequence: list[str] = None
form_buttons: list[list[FieldEditButton | CommandButton | InlineButton]] = None
show_edit_button: bool = True
show_delete_button: bool = True
@dataclass(kw_only=True)
class _BaseFieldDescriptor:
icon: str = None
caption: str | LazyProxy | EntityFieldCaptionCallable | None = None
description: str | LazyProxy | EntityFieldCaptionCallable | None = None
edit_prompt: str | LazyProxy | EntityFieldCaptionCallable | None = None
caption_value: EntityFieldCaptionCallable | None = None
is_visible: bool = True
localizable: bool = False
bool_false_value: str | LazyProxy = "no"
bool_true_value: str | LazyProxy = "yes"
ep_form: str | None = None
ep_parent_field: str | None = None
ep_child_field: str | None = None
dt_type: Literal["date", "datetime"] = "date"
default: Any = None
@dataclass(kw_only=True)
class EntityField(_BaseFieldDescriptor):
name: str | None = None
sm_descriptor: Any = None
@dataclass(kw_only=True)
class Setting(_BaseFieldDescriptor):
name: str | None = None
@dataclass(kw_only=True)
class FormField(_BaseFieldDescriptor):
name: str | None = None
type_: type
@dataclass(kw_only=True)
class FieldDescriptor(_BaseFieldDescriptor):
name: str
field_name: str
type_: type
type_base: type = None
is_list: bool = False
is_optional: bool = False
entity_descriptor: "EntityDescriptor" = None
command: "BotCommand" = None
def __hash__(self):
return self.name.__hash__()
@dataclass(kw_only=True)
class _BaseEntityDescriptor:
icon: str = "📘"
full_name: str | LazyProxy | EntityCaptionCallable | None = None
full_name_plural: str | LazyProxy | EntityCaptionCallable | None = None
description: str | LazyProxy | EntityCaptionCallable | None = None
item_repr: EntityItemCaptionCallable | None = None
default_list: EntityList = field(default_factory=EntityList)
default_form: EntityForm = field(default_factory=EntityForm)
lists: dict[str, EntityList] = field(default_factory=dict[str, EntityList])
forms: dict[str, EntityForm] = field(default_factory=dict[str, EntityForm])
show_in_entities_menu: bool = True
ownership_fields: dict[RoleBase, str] = field(default_factory=dict[RoleBase, str])
permissions: dict[EntityPermission, list[RoleBase]] = field(
default_factory=lambda: {
EntityPermission.LIST: [RoleBase.DEFAULT_USER, RoleBase.SUPER_USER],
EntityPermission.READ: [RoleBase.DEFAULT_USER, RoleBase.SUPER_USER],
EntityPermission.CREATE: [RoleBase.DEFAULT_USER, RoleBase.SUPER_USER],
EntityPermission.UPDATE: [RoleBase.DEFAULT_USER, RoleBase.SUPER_USER],
EntityPermission.DELETE: [RoleBase.DEFAULT_USER, RoleBase.SUPER_USER],
EntityPermission.LIST_ALL: [RoleBase.SUPER_USER],
EntityPermission.READ_ALL: [RoleBase.SUPER_USER],
EntityPermission.CREATE_ALL: [RoleBase.SUPER_USER],
EntityPermission.UPDATE_ALL: [RoleBase.SUPER_USER],
EntityPermission.DELETE_ALL: [RoleBase.SUPER_USER],
}
)
@dataclass(kw_only=True)
class Entity(_BaseEntityDescriptor):
name: str | None = None
@dataclass
class EntityDescriptor(_BaseEntityDescriptor):
name: str
class_name: str
type_: type["BotEntity"]
fields_descriptors: dict[str, FieldDescriptor]
@dataclass(kw_only=True)
class CommandCallbackContext[UT: UserBase]:
keyboard_builder: InlineKeyboardBuilder = field(
default_factory=InlineKeyboardBuilder
)
message_text: str | None = None
register_navigation: bool = True
message: Message | CallbackQuery
callback_data: ContextData
db_session: AsyncSession
user: UT
app: "QBotApp"
state_data: dict[str, Any]
state: FSMContext
form_data: dict[str, Any]
i18n: I18n
kwargs: dict[str, Any] = field(default_factory=dict)
@dataclass(kw_only=True)
class BotCommand:
name: str
caption: str | dict[str, str] | None = None
pre_check: Callable[[Union[Message, CallbackQuery], Any], bool] | None = None
show_in_bot_commands: bool = False
register_navigation: bool = True
clear_navigation: bool = False
clear_state: bool = True
param_form: dict[str, FieldDescriptor] | None = None
show_cancel_in_param_form: bool = True
show_back_in_param_form: bool = True
handler: Callable[[CommandCallbackContext], None]

View File

@@ -0,0 +1,7 @@
from .descriptors import EntityDescriptor
from ._singleton import Singleton
class EntityMetadata(metaclass=Singleton):
def __init__(self):
self.entity_descriptors: dict[str, EntityDescriptor] = {}

View File

@@ -0,0 +1,7 @@
from sqlmodel import SQLModel, Field
class FSMStorage(SQLModel, table=True):
__tablename__ = "fsm_storage"
key: str = Field(primary_key=True)
value: str | None = None

View File

@@ -0,0 +1,5 @@
from .bot_enum import BotEnum, EnumMember
class LanguageBase(BotEnum):
EN = EnumMember("en", {"en": "🇬🇧 english"})

6
src/qbot/model/role.py Normal file
View File

@@ -0,0 +1,6 @@
from .bot_enum import BotEnum, EnumMember
class RoleBase(BotEnum):
SUPER_USER = EnumMember("super_user")
DEFAULT_USER = EnumMember("default_user")

284
src/qbot/model/settings.py Normal file
View File

@@ -0,0 +1,284 @@
from types import NoneType, UnionType
from aiogram.utils.i18n.context import get_i18n
from datetime import datetime
from sqlmodel import SQLModel, Field, select
from typing import Any, get_args, get_origin
from ..db import async_session
from .role import RoleBase
from .descriptors import FieldDescriptor, Setting
from ..utils.serialization import deserialize, serialize
import ujson as json
class DbSettings(SQLModel, table=True):
__tablename__ = "settings"
name: str = Field(primary_key=True)
value: str
class SettingsMetaclass(type):
def __new__(cls, class_name, base_classes, attributes):
settings_descriptors = {}
if base_classes:
settings_descriptors = base_classes[0].__dict__.get(
"_settings_descriptors", {}
)
for annotation in attributes.get("__annotations__", {}):
if annotation in ["_settings_descriptors", "_cache", "_cached_settings"]:
continue
attr_value = attributes.get(annotation)
name = annotation
type_ = attributes["__annotations__"][annotation]
if isinstance(attr_value, Setting):
descriptor_kwargs = attr_value.__dict__.copy()
name = descriptor_kwargs.pop("name") or annotation
attributes[annotation] = FieldDescriptor(
name=name,
field_name=annotation,
type_=type_,
type_base=type_,
**descriptor_kwargs,
)
else:
attributes[annotation] = FieldDescriptor(
name=annotation,
field_name=annotation,
type_=type_,
type_base=type_,
default=attr_value,
)
type_origin = get_origin(type_)
if type_origin is list:
attributes[annotation].is_list = True
attributes[annotation].type_base = type_ = get_args(type_)[0]
elif type_origin == UnionType and get_args(type_)[1] == NoneType:
attributes[annotation].is_optional = True
attributes[annotation].type_base = type_ = get_args(type_)[0]
settings_descriptors[name] = attributes[annotation]
if (
base_classes
and base_classes[0].__name__ == "Settings"
and hasattr(base_classes[0], annotation)
):
setattr(base_classes[0], annotation, attributes[annotation])
attributes["__annotations__"] = {}
attributes["_settings_descriptors"] = settings_descriptors
return super().__new__(cls, class_name, base_classes, attributes)
class Settings(metaclass=SettingsMetaclass):
_cache: dict[str, Any] = dict[str, Any]()
_settings_descriptors: dict[str, FieldDescriptor] = {}
PAGE_SIZE: int = Setting(
default=10,
)
SECURITY_PARAMETERS_ROLES: list[RoleBase] = Setting(
name="SECPARAMS_ROLES", default=[RoleBase.SUPER_USER], is_visible=False
)
APP_STRINGS_WELCOME_P_NAME: str = Setting(
name="AS_WELCOME", default="Welcome, {name}", is_visible=False
)
APP_STRINGS_GREETING_P_NAME: str = Setting(
name="AS_GREETING", default="Hello, {name}", is_visible=False
)
APP_STRINGS_INTERNAL_ERROR_P_ERROR: str = Setting(
name="AS_INTERNAL_ERROR", default="Internal error\n{error}", is_visible=False
)
APP_STRINGS_USER_BLOCKED_P_NAME: str = Setting(
name="AS_USER_BLOCKED", default="User {name} is blocked", is_visible=False
)
APP_STRINGS_FORBIDDEN: str = Setting(
name="AS_FORBIDDEN", default="Forbidden", is_visible=False
)
APP_STRINGS_NOT_FOUND: str = Setting(
name="AS_NOT_FOUND", default="Object not found", is_visible=False
)
APP_STRINGS_MAIN_NENU: str = Setting(
name="AS_MAIN_MENU", default="Main menu", is_visible=False
)
APP_STRINGS_REFERENCES: str = Setting(
name="AS_REFERENCES", default="References", is_visible=False
)
APP_STRINGS_REFERENCES_BTN: str = Setting(
name="AS_REFERENCES_BTN", default="📚 References", is_visible=False
)
APP_STRINGS_SETTINGS: str = Setting(
name="AS_SETTINGS", default="Settings", is_visible=False
)
APP_STRINGS_SETTINGS_BTN: str = Setting(
name="AS_SETTINGS_BTN", default="⚙️ Settings", is_visible=False
)
APP_STRINGS_PARAMETERS: str = Setting(
name="AS_PARAMETERS", default="Parameters", is_visible=False
)
APP_STRINGS_PARAMETERS_BTN: str = Setting(
name="AS_PARAMETERS_BTN", default="🎛️ Parameters", is_visible=False
)
APP_STRINGS_LANGUAGE: str = Setting(
name="AS_LANGUAGE", default="Language", is_visible=False
)
APP_STRINGS_LANGUAGE_BTN: str = Setting(
name="AS_LANGUAGE_BTN", default="🗣️ Language", is_visible=False
)
APP_STRINGS_BACK_BTN: str = Setting(
name="AS_BACK_BTN", default="⬅️ Back", is_visible=False
)
APP_STRINGS_DELETE_BTN: str = Setting(
name="AS_DELETE_BTN", default="🗑️ Delete", is_visible=False
)
APP_STRINGS_CONFIRM_DELETE_P_NAME: str = Setting(
name="AS_CONFIRM_DEL",
default='Are you sure you want to delete "{name}"?',
is_visible=False,
)
APP_STRINGS_EDIT_BTN: str = Setting(
name="AS_EDIT_BTN", default="✏️ Edit", is_visible=False
)
APP_STRINGS_ADD_BTN: str = Setting(
name="AS_ADD_BTN", default=" Add", is_visible=False
)
APP_STRINGS_YES_BTN: str = Setting(
name="AS_YES_BTN", default="✅ Yes", is_visible=False
)
APP_STRINGS_NO_BTN: str = Setting(
name="AS_NO_BTN", default="❌ No", is_visible=False
)
APP_STRINGS_CANCEL_BTN: str = Setting(
name="AS_CANCEL_BTN", default="❌ Cancel", is_visible=False
)
APP_STRINGS_CLEAR_BTN: str = Setting(
name="AS_CLEAR_BTN", default="⌫ Clear", is_visible=False
)
APP_STRINGS_DONE_BTN: str = Setting(
name="AS_DONE_BTN", default="✅ Done", is_visible=False
)
APP_STRINGS_SKIP_BTN: str = Setting(
name="AS_SKIP_BTN", default="⏩️ Skip", is_visible=False
)
APP_STRINGS_FIELD_EDIT_PROMPT_TEMPLATE_P_NAME_VALUE: str = Setting(
name="AS_FIELDEDIT_PROMPT",
default='Enter new value for "{name}" (current value: {value})',
is_visible=False,
)
APP_STRINGS_FIELD_CREATE_PROMPT_TEMPLATE_P_NAME: str = Setting(
name="AS_FIELDCREATE_PROMPT",
default='Enter new value for "{name}"',
is_visible=False,
)
APP_STRINGS_STRING_EDITOR_LOCALE_TEMPLATE_P_NAME: str = Setting(
name="AS_STREDIT_LOC_TEMPLATE", default='string for "{name}"', is_visible=False
)
APP_STRINGS_VIEW_FILTER_EDIT_PROMPT: str = Setting(
name="AS_FILTEREDIT_PROMPT", default="Enter filter value", is_visible=False
)
APP_STRINGS_INVALID_INPUT: str = Setting(
name="AS_INVALID_INPUT", default="Invalid input", is_visible=False
)
@classmethod
async def get[T](cls, param: T, all_locales=False, locale: str = None) -> T:
name = param.field_name
if name not in cls._cache.keys():
cls._cache[name] = await cls.load_param(param)
ret_val = cls._cache[name]
if param.localizable and not all_locales:
if not locale:
locale = get_i18n().current_locale
try:
obj = json.loads(ret_val)
except Exception:
return ret_val
return obj.get(locale, obj[list(obj.keys())[0]])
return ret_val
@classmethod
async def load_param(cls, param: FieldDescriptor) -> Any:
async with async_session() as session:
db_setting = (
await session.exec(
select(DbSettings).where(DbSettings.name == param.field_name)
)
).first()
if db_setting:
return await deserialize(
session=session, type_=param.type_, value=db_setting.value
)
return (
param.default
if param.default
else (
[]
if (get_origin(param.type_) is list or param.type_ is list)
else datetime(2000, 1, 1)
if param.type_ == datetime
else param.type_()
)
)
@classmethod
async def load_params(cls):
async with async_session() as session:
db_settings = (await session.exec(select(DbSettings))).all()
for db_setting in db_settings:
if db_setting.name in cls.__dict__:
setting = cls.__dict__[db_setting.name] # type: FieldDescriptor
cls._cache[db_setting.name] = await deserialize(
session=session,
type_=setting.type_,
value=db_setting.value,
default=setting.default,
)
cls._loaded = True
@classmethod
async def set_param(cls, param: str | FieldDescriptor, value) -> None:
if isinstance(param, str):
param = cls._settings_descriptors[param]
ser_value = serialize(value, param)
async with async_session() as session:
db_setting = (
await session.exec(
select(DbSettings).where(DbSettings.name == param.field_name)
)
).first()
if db_setting is None:
db_setting = DbSettings(name=param.field_name)
db_setting.value = str(ser_value)
session.add(db_setting)
await session.commit()
cls._cache[param.field_name] = value
@classmethod
def list_params(cls) -> dict[str, FieldDescriptor]:
return cls._settings_descriptors
@classmethod
async def get_params(cls) -> dict[FieldDescriptor, Any]:
params = cls.list_params()
return {
param: await cls.get(param, all_locales=True) for _, param in params.items()
}

23
src/qbot/model/user.py Normal file
View File

@@ -0,0 +1,23 @@
from sqlmodel import Field, ARRAY
from .bot_entity import BotEntity
from .bot_enum import EnumType
from .language import LanguageBase
from .role import RoleBase
from .settings import DbSettings as DbSettings
from .fsm_storage import FSMStorage as FSMStorage
from .view_setting import ViewSetting as ViewSetting
class UserBase(BotEntity, table=False):
__tablename__ = "user"
lang: LanguageBase = Field(sa_type=EnumType(LanguageBase), default=LanguageBase.EN)
is_active: bool = True
name: str
roles: list[RoleBase] = Field(
sa_type=ARRAY(EnumType(RoleBase)), default=[RoleBase.DEFAULT_USER]
)

View File

@@ -0,0 +1,39 @@
from sqlmodel import SQLModel, Field, BigInteger
from sqlalchemy.ext.asyncio.session import AsyncSession
from . import session_dep
class ViewSetting(SQLModel, table=True):
__tablename__ = "view_setting"
user_id: int = Field(
sa_type=BigInteger, primary_key=True, foreign_key="user.id", ondelete="CASCADE"
)
entity_name: str = Field(primary_key=True)
filter: str | None = None
@classmethod
@session_dep
async def get_filter(
cls, *, session: AsyncSession | None = None, user_id: int, entity_name: str
):
setting = await session.get(cls, (user_id, entity_name))
return setting.filter if setting else None
@classmethod
@session_dep
async def set_filter(
cls,
*,
session: AsyncSession | None = None,
user_id: int,
entity_name: str,
filter: str,
):
setting = await session.get(cls, (user_id, entity_name))
if setting:
setting.filter = filter
else:
setting = cls(user_id=user_id, entity_name=entity_name, filter=filter)
session.add(setting)
await session.commit()

79
src/qbot/router.py Normal file
View File

@@ -0,0 +1,79 @@
from functools import wraps
from types import UnionType
from typing import Callable, Union, get_args, get_origin
from .model.descriptors import (
BotCommand,
CommandCallbackContext,
FieldDescriptor,
FormField,
)
class Router:
def __init__(self):
self._commands = dict[str, BotCommand]()
def command(
self,
name: str,
caption: str | dict[str, str] | None = None,
pre_check: Callable[[CommandCallbackContext], bool] | None = None,
show_in_bot_commands: bool = False,
register_navigation: bool = True,
clear_navigation: bool = False,
clear_state: bool = True,
show_cancel_in_param_form: bool = True,
show_back_in_param_form: bool = True,
form_fields: list[FormField] = list[FormField](),
):
def decorator(func: Callable[[CommandCallbackContext], None]):
form_fields_dict = dict[str, FieldDescriptor]()
for field in form_fields:
is_list = False
is_optional = False
type_origin = get_origin(field.type_)
if type_origin is list:
is_list = True
type_base = get_args(field.type_)[0]
elif type_origin in [Union, UnionType] and type(None) in get_args(
field.type_
):
is_optional = True
type_base = get_args(field.type_)[0]
else:
type_base = field.type_
form_fields_dict[field.name] = FieldDescriptor(
field_name=field.name,
type_base=type_base,
is_list=is_list,
is_optional=is_optional,
**field.__dict__,
)
cmd = BotCommand(
name=name,
caption=caption,
pre_check=pre_check,
show_in_bot_commands=show_in_bot_commands,
register_navigation=register_navigation,
clear_navigation=clear_navigation,
clear_state=clear_state,
param_form=form_fields_dict,
show_cancel_in_param_form=show_cancel_in_param_form,
show_back_in_param_form=show_back_in_param_form,
handler=func,
)
for field in form_fields_dict.values():
field.command = cmd
self._commands[cmd.name] = cmd
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
return decorator

214
src/qbot/utils/main.py Normal file
View File

@@ -0,0 +1,214 @@
from babel.support import LazyProxy
from inspect import signature
from aiogram.types import Message, CallbackQuery
from aiogram.utils.i18n import I18n
from typing import Any, TYPE_CHECKING
import ujson as json
from ..model.bot_entity import BotEntity
from ..model.bot_enum import BotEnum
from ..model.settings import Settings
from ..model.descriptors import (
FieldDescriptor,
EntityDescriptor,
EntityItemCaptionCallable,
EntityFieldCaptionCallable,
EntityPermission,
EntityCaptionCallable,
)
from ..bot.handlers.context import ContextData, CommandContext
if TYPE_CHECKING:
from ..model.user import UserBase
from ..main import QBotApp
def get_user_permissions(
user: "UserBase", entity_descriptor: EntityDescriptor
) -> list[EntityPermission]:
permissions = list[EntityPermission]()
for permission, roles in entity_descriptor.permissions.items():
for role in roles:
if role in user.roles:
permissions.append(permission)
break
return permissions
def get_local_text(text: str, locale: str = None) -> str:
if not locale:
i18n = I18n.get_current(no_error=True)
if i18n:
locale = i18n.current_locale
else:
locale = "en"
try:
obj = json.loads(text) # @IgnoreException
except Exception:
return text
else:
return obj.get(locale, obj[list(obj.keys())[0]])
def check_entity_permission(
entity: BotEntity, user: "UserBase", permission: EntityPermission
) -> bool:
perm_mapping = {
EntityPermission.LIST: EntityPermission.LIST_ALL,
EntityPermission.READ: EntityPermission.READ_ALL,
EntityPermission.UPDATE: EntityPermission.UPDATE_ALL,
EntityPermission.CREATE: EntityPermission.CREATE_ALL,
EntityPermission.DELETE: EntityPermission.DELETE_ALL,
}
if permission not in perm_mapping:
raise ValueError(f"Invalid permission: {permission}")
entity_descriptor = entity.__class__.bot_entity_descriptor
permissions = get_user_permissions(user, entity_descriptor)
if perm_mapping[permission] in permissions:
return True
ownership_filds = entity_descriptor.ownership_fields
for role in user.roles:
if role in ownership_filds:
if getattr(entity, ownership_filds[role]) == user.id:
return True
else:
if permission in permissions:
return True
return False
def get_send_message(message: Message | CallbackQuery):
if isinstance(message, Message):
return message.answer
else:
return message.message.edit_text
def clear_state(state_data: dict, clear_nav: bool = False):
if clear_nav:
state_data.clear()
else:
stack = state_data.get("navigation_stack")
context = state_data.get("navigation_context")
state_data.clear()
if stack:
state_data["navigation_stack"] = stack
if context:
state_data["navigation_context"] = context
def get_entity_item_repr(
entity: BotEntity, item_repr: EntityItemCaptionCallable | None = None
) -> str:
descr = entity.bot_entity_descriptor
if item_repr:
return item_repr(descr, entity)
return (
descr.item_repr(descr, entity)
if descr.item_repr
else f"{
get_callable_str(descr.full_name, descr, entity)
if descr.full_name
else descr.name
}: {str(entity.id)}"
)
def get_value_repr(
value: Any, field_descriptor: FieldDescriptor, locale: str | None = None
) -> str:
if value is None:
return ""
type_ = field_descriptor.type_base
if isinstance(value, bool):
return "【✔︎】" if value else "【 】"
elif field_descriptor.is_list:
if issubclass(type_, BotEntity):
return f"[{', '.join([get_entity_item_repr(item) for item in value])}]"
elif issubclass(type_, BotEnum):
return f"[{', '.join(item.localized(locale) for item in value)}]"
elif type_ is str:
return f"[{', '.join([f'"{item}"' for item in value])}]"
else:
return f"[{', '.join([str(item) for item in value])}]"
elif issubclass(type_, BotEntity):
return get_entity_item_repr(value)
elif issubclass(type_, BotEnum):
return value.localized(locale)
elif isinstance(value, str):
if field_descriptor and field_descriptor.localizable:
return get_local_text(text=value, locale=locale)
return value
elif isinstance(value, int):
return str(value)
elif isinstance(value, float):
return str(value)
else:
return str(value)
def get_callable_str(
callable_str: (
str
| LazyProxy
| EntityCaptionCallable
| EntityItemCaptionCallable
| EntityFieldCaptionCallable
),
descriptor: FieldDescriptor | EntityDescriptor,
entity: Any = None,
value: Any = None,
) -> str:
if isinstance(callable_str, str):
return callable_str
elif isinstance(callable_str, LazyProxy):
return callable_str.value
elif callable(callable_str):
args = signature(callable_str).parameters
if len(args) == 1:
return callable_str(descriptor)
elif len(args) == 2:
return callable_str(descriptor, entity)
elif len(args) == 3:
return callable_str(descriptor, entity, value)
def get_entity_descriptor(
app: "QBotApp", callback_data: ContextData
) -> EntityDescriptor:
if callback_data.entity_name:
return app.entity_metadata.entity_descriptors[callback_data.entity_name]
return None
def get_field_descriptor(
app: "QBotApp", callback_data: ContextData
) -> FieldDescriptor | None:
if callback_data.context == CommandContext.SETTING_EDIT:
return Settings.list_params()[callback_data.field_name]
elif callback_data.context == CommandContext.COMMAND_FORM:
command = app.bot_commands[callback_data.user_command.split("&")[0]]
if (
command
and command.param_form
and callback_data.field_name in command.param_form
):
return command.param_form[callback_data.field_name]
elif callback_data.context in [
CommandContext.ENTITY_CREATE,
CommandContext.ENTITY_EDIT,
CommandContext.ENTITY_FIELD_EDIT,
]:
entity_descriptor = get_entity_descriptor(app, callback_data)
if entity_descriptor:
return entity_descriptor.fields_descriptors.get(callback_data.field_name)
return None

View File

@@ -0,0 +1,50 @@
from ..bot.handlers.context import ContextData, CallbackCommand
def save_navigation_context(
callback_data: ContextData, state_data: dict
) -> list[ContextData]:
stack = [
ContextData.unpack(item) for item in state_data.get("navigation_stack", [])
]
data_nc = state_data.get("navigation_context")
navigation_context = ContextData.unpack(data_nc) if data_nc else None
if callback_data.back:
callback_data.back = False
if stack:
stack.pop()
else:
if (
stack
and navigation_context
and navigation_context.command == callback_data.command
and navigation_context.entity_name == callback_data.entity_name
and navigation_context.entity_id == callback_data.entity_id
and navigation_context.command != CallbackCommand.USER_COMMAND
):
navigation_context = callback_data
elif navigation_context:
stack.append(navigation_context)
state_data["navigation_stack"] = [item.pack() for item in stack]
state_data["navigation_context"] = callback_data.pack()
return stack
def pop_navigation_context(stack: list[ContextData]) -> ContextData | None:
if stack:
data = stack[-1]
data.back = True
return data
def get_navigation_context(
state_data: dict,
) -> tuple[list[ContextData], ContextData | None]:
data_nc = state_data.get("navigation_context")
context = ContextData.unpack(data_nc) if data_nc else None
return (
[ContextData.unpack(item) for item in state_data.get("navigation_stack", [])],
context,
)

View File

@@ -0,0 +1,90 @@
from datetime import datetime, time
from decimal import Decimal
from sqlmodel import select, column
from sqlmodel.ext.asyncio.session import AsyncSession
from typing import Any, Union, get_origin, get_args
from types import UnionType, NoneType
import ujson as json
from ..model.bot_entity import BotEntity
from ..model.bot_enum import BotEnum
from ..model.descriptors import FieldDescriptor
async def deserialize[T](session: AsyncSession, type_: type[T], value: str = None) -> T:
type_origin = get_origin(type_)
is_optional = False
if type_origin in [UnionType, Union]:
args = get_args(type_)
if args[1] is NoneType:
type_ = args[0]
if value is None:
return None
is_optional = True
if get_origin(type_) is list:
arg_type = None
args = get_args(type_)
if args:
arg_type = args[0]
values = json.loads(value) if value else []
if arg_type:
if issubclass(arg_type, BotEntity):
ret = list[arg_type]()
items = (
await session.exec(select(arg_type).where(column("id").in_(values)))
).all()
for item in items:
ret.append(item)
return ret
elif issubclass(arg_type, BotEnum):
return [arg_type(value) for value in values]
else:
return [arg_type(value) for value in values]
else:
return values
elif issubclass(type_, BotEntity):
if is_optional and not value:
return None
return await session.get(type_, int(value))
elif issubclass(type_, BotEnum):
if is_optional and not value:
return None
return type_(value)
elif type_ is time:
if is_optional and not value:
return None
return time.fromisoformat(value.replace("-", ":"))
elif type_ is datetime:
if is_optional and not value:
return None
if value[-3] == "-":
return datetime.strptime(value, "%Y-%m-%d %H-%M")
else:
return datetime.fromisoformat(value)
elif type_ is bool:
return value == "True"
elif type_ is Decimal:
if is_optional and not value:
return None
return Decimal(value)
if is_optional and not value:
return None
return type_(value)
def serialize(value: Any, field_descriptor: FieldDescriptor) -> str:
if value is None:
return ""
type_ = field_descriptor.type_base
if field_descriptor.is_list:
if issubclass(type_, BotEntity):
return json.dumps([item.id for item in value], ensure_ascii=False)
elif issubclass(type_, BotEnum):
return json.dumps([item.value for item in value], ensure_ascii=False)
else:
return json.dumps(value, ensure_ascii=False)
elif issubclass(type_, BotEntity):
return str(value.id) if value else ""
return str(value)