BotForge/quickbot
2
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

upd defult editing field sequences generation when ownership fields defined

Browse Source
This commit is contained in:
Alexander Kalinovsky
2025-03-13 16:52:03 +07:00
parent 02aec23b84
commit f0db2b2830
17 changed files with 223 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/qbot/bot/handlers/editors/boolean.py
View File

@@ -6,6 +6,7 @@ from babel.support import LazyProxy
from logging import getLogger
from ....model.descriptors import FieldDescriptor
from ....model.user import UserBase
from ..context import ContextData, CallbackCommand
from ....utils.main import get_send_message
from .wrapper import wrap_editor
@@ -20,6 +21,7 @@ async def bool_editor(
edit_prompt: str,
field_descriptor: FieldDescriptor,
callback_data: ContextData,
user: UserBase,
**kwargs,
):
keyboard_builder = InlineKeyboardBuilder()
@@ -70,6 +72,7 @@ async def bool_editor(
field_descriptor=field_descriptor,
callback_data=callback_data,
state_data=state_data,
user=user,
)
state: FSMContext = kwargs["state"]

7
src/qbot/bot/handlers/editors/date.py
View File

@@ -8,6 +8,7 @@ from typing import TYPE_CHECKING
from ....model.descriptors import FieldDescriptor
from ....model.settings import Settings
from ....model.user import UserBase
from ..context import ContextData, CallbackCommand
from ....utils.main import get_send_message, get_field_descriptor
from .wrapper import wrap_editor
@@ -49,6 +50,7 @@ async def time_picker(
callback_data: ContextData,
current_value: datetime | time,
state: FSMContext,
user: UserBase,
edit_prompt: str | None = None,
**kwargs,
):
@@ -162,6 +164,7 @@ async def time_picker(
field_descriptor=field_descriptor,
callback_data=callback_data,
state_data=state_data,
user=user,
)
await state.set_data(state_data)
@@ -179,6 +182,7 @@ async def date_picker(
callback_data: ContextData,
current_value: datetime,
state: FSMContext,
user: UserBase,
edit_prompt: str | None = None,
**kwargs,
):
@@ -273,6 +277,7 @@ async def date_picker(
field_descriptor=field_descriptor,
callback_data=callback_data,
state_data=state_data,
user=user,
)
await state.set_data(state_data)
@@ -292,6 +297,7 @@ async def date_picker_year(
callback_data: ContextData,
app: "QBotApp",
state: FSMContext,
user: UserBase,
**kwargs,
):
start_date = datetime.strptime(callback_data.data, "%Y-%m-%d %H-%M")
@@ -365,6 +371,7 @@ async def date_picker_year(
field_descriptor=field_descriptor,
callback_data=callback_data,
state_data=state_data,
user=user,
)
await query.message.edit_reply_markup(reply_markup=keyboard_builder.as_markup())

1
src/qbot/bot/handlers/editors/entity.py
View File

@@ -289,6 +289,7 @@ async def render_entity_picker(
field_descriptor=field_descriptor,
callback_data=callback_data,
state_data=state_data,
user=user,
)
await state.set_data(state_data)

2
src/qbot/bot/handlers/editors/main.py
View File

@@ -107,6 +107,8 @@ async def field_editor(message: Message | CallbackQuery, **kwargs):
await db_session.commit()
stack, context = get_navigation_context(state_data=state_data)
kwargs.update({"callback_data": context})
return await entity_item(
query=message, navigation_stack=stack, **kwargs
)

49
src/qbot/bot/handlers/editors/main_callbacks.py
View File

@@ -22,6 +22,7 @@ from ....utils.main import (
clear_state,
get_entity_descriptor,
get_field_descriptor,
build_field_sequence,
)
from ....utils.serialization import deserialize
from ..common.routing import route_callback
@@ -168,7 +169,15 @@ async def process_field_edit_callback(message: Message | CallbackQuery, **kwargs
form_name, entity_descriptor.default_form
)
field_sequence = form.edit_field_sequence
if form.edit_field_sequence:
field_sequence = form.edit_field_sequence
else:
field_sequence = build_field_sequence(
entity_descriptor=entity_descriptor,
user=user,
callback_data=callback_data,
)
current_index = (
field_sequence.index(callback_data.field_name)
if callback_data.context
@@ -251,10 +260,14 @@ async def process_field_edit_callback(message: Message | CallbackQuery, **kwargs
# 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
user_permissions = get_user_permissions(user, entity_descriptor)
# 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
for role in user.roles:
if (
role in entity_descriptor.ownership_fields
and EntityPermission.CREATE_ALL not in user_permissions
):
entity_data[entity_descriptor.ownership_fields[role]] = user.id
deser_entity_data = {
key: await deserialize(
@@ -284,9 +297,19 @@ async def process_field_edit_callback(message: Message | CallbackQuery, **kwargs
if entity_descriptor.on_created:
if iscoroutinefunction(entity_descriptor.on_created):
await entity_descriptor.on_created(new_entity, EntityEventContext(db_session=db_session, app=app))
await entity_descriptor.on_created(
new_entity,
EntityEventContext(
db_session=db_session, app=app, message=message
),
)
else:
entity_descriptor.on_created(new_entity, EntityEventContext(db_session=db_session, app=app))
entity_descriptor.on_created(
new_entity,
EntityEventContext(
db_session=db_session, app=app, message=message
),
)
form_name = (
callback_data.form_params.split("&")[0]
@@ -336,9 +359,19 @@ async def process_field_edit_callback(message: Message | CallbackQuery, **kwargs
if entity_descriptor.on_updated:
if iscoroutinefunction(entity_descriptor.on_updated):
await entity_descriptor.on_updated(entity, EntityEventContext(db_session=db_session, app=app))
await entity_descriptor.on_updated(
entity,
EntityEventContext(
db_session=db_session, app=app, message=message
),
)
else:
entity_descriptor.on_updated(entity, EntityEventContext(db_session=db_session, app=app))
entity_descriptor.on_updated(
entity,
EntityEventContext(
db_session=db_session, app=app, message=message
),
)
elif callback_data.context == CommandContext.COMMAND_FORM:
clear_state(state_data=state_data)

3
src/qbot/bot/handlers/editors/string.py
View File

@@ -8,6 +8,7 @@ from typing import Any
from ....model.descriptors import FieldDescriptor
from ....model.language import LanguageBase
from ....model.settings import Settings
from ....model.user import UserBase
from ....utils.main import get_send_message, get_local_text
from ....utils.serialization import serialize
from ..context import ContextData, CallbackCommand
@@ -25,6 +26,7 @@ async def string_editor(
current_value: Any,
edit_prompt: str,
state: FSMContext,
user: UserBase,
locale_index: int = 0,
**kwargs,
):
@@ -90,6 +92,7 @@ async def string_editor(
field_descriptor=field_descriptor,
callback_data=callback_data,
state_data=state_data,
user=user,
)
await state.set_data(state_data)

12
src/qbot/bot/handlers/editors/wrapper.py
View File

@@ -3,8 +3,10 @@ from aiogram.utils.keyboard import InlineKeyboardBuilder
from ....model.settings import Settings
from ....model.descriptors import FieldDescriptor
from ....model.user import UserBase
from ..context import ContextData, CallbackCommand, CommandContext
from ....utils.navigation import get_navigation_context, pop_navigation_context
from ....utils.main import build_field_sequence
async def wrap_editor(
@@ -12,6 +14,7 @@ async def wrap_editor(
field_descriptor: FieldDescriptor,
callback_data: ContextData,
state_data: dict,
user: UserBase,
):
if callback_data.context in [
CommandContext.ENTITY_CREATE,
@@ -36,7 +39,14 @@ async def wrap_editor(
form = field_descriptor.entity_descriptor.forms.get(
form_name, field_descriptor.entity_descriptor.default_form
)
field_sequence = form.edit_field_sequence
if form.edit_field_sequence:
field_sequence = form.edit_field_sequence
else:
field_sequence = build_field_sequence(
entity_descriptor=field_descriptor.entity_descriptor,
user=user,
callback_data=callback_data,
)
field_index = (
field_sequence.index(field_descriptor.name)
if callback_data.context

91
src/qbot/bot/handlers/forms/entity_form.py
View File

@@ -1,3 +1,4 @@
from inspect import iscoroutinefunction
from typing import TYPE_CHECKING
from aiogram import Router, F
from aiogram.fsm.context import FSMContext
@@ -6,7 +7,12 @@ 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.descriptors import (
FieldEditButton,
CommandButton,
InlineButton,
EntityEventContext,
)
from ....model.settings import Settings
from ....model.user import UserBase
from ....model import EntityPermission
@@ -17,6 +23,8 @@ from ....utils.main import (
get_value_repr,
get_callable_str,
get_entity_descriptor,
build_field_sequence,
get_user_permissions,
)
from ..context import ContextData, CallbackCommand, CommandContext
from ....utils.navigation import (
@@ -89,10 +97,11 @@ async def entity_item(
)
if form.form_buttons:
context = EntityEventContext(db_session=db_session, app=app, message=query)
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):
if button.visibility and not button.visibility(entity_item, context):
continue
if isinstance(button, FieldEditButton) and can_edit:
@@ -157,7 +166,10 @@ async def entity_item(
if isinstance(button.command, ContextData):
btn_cdata = button.command
elif callable(button.command):
btn_cdata = button.command(callback_data, entity_item)
if iscoroutinefunction(button.command):
btn_cdata = await button.command(entity_item, context)
else:
btn_cdata = button.command(entity_item, context)
elif isinstance(button.command, str):
btn_cdata = ContextData(
command=CallbackCommand.USER_COMMAND,
@@ -175,13 +187,26 @@ async def entity_item(
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 iscoroutinefunction(button.inline_button):
btn_row.append(
await button.inline_button(entity_item, context)
)
else:
btn_row.append(button.inline_button(entity_item, context))
if btn_row:
keyboard_builder.row(*btn_row)
edit_delete_row = []
if can_edit and form.show_edit_button:
if form.edit_field_sequence:
field_sequence = form.edit_field_sequence
else:
field_sequence = build_field_sequence(
entity_descriptor=entity_descriptor,
user=user,
callback_data=callback_data,
)
edit_delete_row.append(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_EDIT_BTN)),
@@ -191,7 +216,7 @@ async def entity_item(
entity_name=entity_descriptor.name,
entity_id=str(entity_item.id),
form_params=callback_data.form_params,
field_name=form.edit_field_sequence[0],
field_name=field_sequence[0],
).pack(),
)
)
@@ -238,25 +263,47 @@ async def entity_item(
item_text = f"<b><u><i>{entity_caption or entity_descriptor.name}:</i></u></b> <b>{entity_item_repr}</b>"
user_permissions = get_user_permissions(user, entity_descriptor)
for field_descriptor in entity_descriptor.fields_descriptors.values():
if field_descriptor.is_visible:
field_caption = await get_callable_str(
field_descriptor.caption, field_descriptor, entity_item
if (
field_descriptor.is_visible is not None
and not field_descriptor.is_visible
):
continue
skip = False
for own_field in entity_descriptor.ownership_fields.items():
if (
own_field[1].rstrip("_id")
== field_descriptor.field_name.rstrip("_id")
and own_field[0] in user.roles
and EntityPermission.READ_ALL not in user_permissions
):
skip = True
break
if skip:
continue
field_caption = await get_callable_str(
field_descriptor.caption, field_descriptor, entity_item
)
if field_descriptor.caption_value:
value = await get_callable_str(
field_descriptor.caption_value,
field_descriptor,
entity_item,
getattr(entity_item, field_descriptor.field_name),
)
if field_descriptor.caption_value:
value = await get_callable_str(
field_descriptor.caption_value,
field_descriptor,
entity_item,
getattr(entity_item, field_descriptor.field_name),
)
else:
value = await 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 ''}"
else:
value = await 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:

12
src/qbot/bot/handlers/forms/entity_form_callbacks.py
View File

@@ -50,17 +50,21 @@ async def entity_delete_callback(query: CallbackQuery, **kwargs):
)
if callback_data.data == "yes":
entity = await entity_descriptor.type_.remove(
session=db_session, id=int(callback_data.entity_id), commit=True
)
if entity_descriptor.on_deleted:
if iscoroutinefunction(entity_descriptor.on_created):
await entity_descriptor.on_deleted(entity, EntityEventContext(db_session=db_session, app=app))
await entity_descriptor.on_deleted(
entity,
EntityEventContext(db_session=db_session, app=app, message=query),
)
else:
entity_descriptor.on_deleted(entity, EntityEventContext(db_session=db_session, app=app))
entity_descriptor.on_deleted(
entity,
EntityEventContext(db_session=db_session, app=app, message=query),
)
await route_callback(message=query, **kwargs)

11
src/qbot/bot/handlers/forms/entity_list.py
View File

@@ -18,6 +18,7 @@ from ....utils.main import (
clear_state,
get_entity_descriptor,
get_callable_str,
build_field_sequence,
)
from ....utils.serialization import deserialize
from ..context import ContextData, CallbackCommand, CommandContext
@@ -116,6 +117,14 @@ async def entity_list(
EntityPermission.CREATE in user_permissions
or EntityPermission.CREATE_ALL in user_permissions
) and form_list.show_add_new_button:
if form_item.edit_field_sequence:
field_sequence = form_item.edit_field_sequence
else:
field_sequence = build_field_sequence(
entity_descriptor=entity_descriptor,
user=user,
callback_data=callback_data,
)
keyboard_builder.row(
InlineKeyboardButton(
text=(await Settings.get(Settings.APP_STRINGS_ADD_BTN)),
@@ -123,7 +132,7 @@ async def entity_list(
command=CallbackCommand.FIELD_EDITOR,
context=CommandContext.ENTITY_CREATE,
entity_name=entity_descriptor.name,
field_name=form_item.edit_field_sequence[0],
field_name=field_sequence[0],
form_params=form_list.item_form,
).pack(),
)

4
src/qbot/bot/handlers/start.py
View File

@@ -21,9 +21,7 @@ async def start(message: Message, **kwargs):
app: QBotApp = kwargs["app"]
if app.start_handler:
await app.start_handler(
default_start_handler, message, **kwargs
)
await app.start_handler(default_start_handler, message, **kwargs)
else:
await default_start_handler(message, **kwargs)

4
src/qbot/config/__init__.py
View File

@@ -38,9 +38,7 @@ class Config(BaseSettings):
def API_URL(self) -> str:
if self.USE_NGROK:
return self.NGROK_URL
return (
f"https://{self.API_DOMAIN}"
)
return f"https://{self.API_DOMAIN}"
API_PORT: int = 8000

14
src/qbot/main.py
View File

@@ -21,12 +21,12 @@ 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()
@@ -44,10 +44,10 @@ async def default_lifespan(app: "QBotApp"):
if app.lifespan_bot_init:
await app.bot_close()
if app.config.USE_NGROK:
app.ngrok_stop()
logger.info("qbot app stopped")
@@ -129,13 +129,11 @@ class QBotApp[UserType: UserBase](FastAPI):
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
@@ -150,7 +148,6 @@ class QBotApp[UserType: UserBase](FastAPI):
pyngrok_config=PyngrokConfig(auth_token=self.config.NGROK_AUTH_TOKEN),
)
self.config.NGROK_URL = tunnel.public_url
def ngrok_stop(self):
try:
@@ -163,10 +160,7 @@ class QBotApp[UserType: UserBase](FastAPI):
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():
@@ -199,7 +193,5 @@ class QBotApp[UserType: UserBase](FastAPI):
secret_token=self.bot_auth_token,
)
async def bot_close(self):
await self.bot.delete_webhook()

38
src/qbot/model/bot_entity.py
View File

@@ -12,6 +12,7 @@ from typing import (
dataclass_transform,
)
from pydantic import BaseModel
from pydantic_core import PydanticUndefined
from sqlmodel import SQLModel, BigInteger, Field, select, func, column, col
from sqlmodel.main import FieldInfo
from sqlmodel.ext.asyncio.session import AsyncSession
@@ -64,7 +65,16 @@ class BotEntityMetaclass(SQLModelMetaclass):
if attribute_value:
if isinstance(attribute_value, EntityField):
descriptor_kwargs = attribute_value.__dict__.copy()
sm_descriptor = descriptor_kwargs.pop("sm_descriptor", None)
sm_descriptor = descriptor_kwargs.pop("sm_descriptor", None) # type: FieldInfo
if attribute_value.default is not None:
if (
sm_descriptor
and sm_descriptor.default is PydanticUndefined
):
sm_descriptor.default = attribute_value.default
else:
sm_descriptor = Field(default=attribute_value.default)
if sm_descriptor:
namespace[annotation] = sm_descriptor
@@ -157,22 +167,22 @@ class BotEntityMetaclass(SQLModelMetaclass):
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")
]
# descriptor_fields_sequence = [
# key
# for key, val in bot_fields_descriptors.items()
# if not (val.is_optional or val.name == "id" or val.name[:-3] == "_id")
# ]
entity_descriptor: EntityDescriptor = namespace["bot_entity_descriptor"]
# 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
)
# 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 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"]

3
src/qbot/model/descriptors.py
View File

@@ -93,7 +93,7 @@ class _BaseFieldDescriptor:
description: str | LazyProxy | EntityFieldCaptionCallable | None = None
edit_prompt: str | LazyProxy | EntityFieldCaptionCallable | None = None
caption_value: EntityFieldCaptionCallable | None = None
is_visible: bool = True
is_visible: bool | None = None
localizable: bool = False
bool_false_value: str | LazyProxy = "no"
bool_true_value: str | LazyProxy = "yes"
@@ -204,6 +204,7 @@ class CommandCallbackContext[UT: UserBase]:
class EntityEventContext:
db_session: AsyncSession
app: "QBotApp"
message: Message | CallbackQuery | None = None
@dataclass(kw_only=True)

37
src/qbot/utils/main.py
View File

@@ -222,3 +222,40 @@ def get_field_descriptor(
if entity_descriptor:
return entity_descriptor.fields_descriptors.get(callback_data.field_name)
return None
def build_field_sequence(
entity_descriptor: EntityDescriptor, user: "UserBase", callback_data: ContextData
):
field_sequence = list[str]()
# exclude ownership fields from edit if user has no CREATE_ALL/UPDATE_ALL permission
user_permissions = get_user_permissions(user, entity_descriptor)
for fd in entity_descriptor.fields_descriptors.values():
if not (
fd.is_optional
or fd.field_name == "id"
or fd.field_name[:-3] == "_id"
or fd.default is not None
):
skip = False
for own_field in entity_descriptor.ownership_fields.items():
if (
own_field[1].rstrip("_id") == fd.field_name.rstrip("_id")
and own_field[0] in user.roles
and (
(
EntityPermission.CREATE_ALL not in user_permissions
and callback_data.context == CommandContext.ENTITY_CREATE
)
or (
EntityPermission.UPDATE_ALL not in user_permissions
and callback_data.context == CommandContext.ENTITY_EDIT
)
)
):
skip = True
break
if not skip:
field_sequence.append(fd.field_name)
return field_sequence

2
tests/bot.py
View File

@@ -21,7 +21,6 @@ from typing import Optional
class User(UserBase):
bot_entity_descriptor = Entity(
icon="👤",
full_name="User",
@@ -56,7 +55,6 @@ class User(UserBase):
class Entity(BotEntity):
bot_entity_descriptor = Entity(
icon="📦",
full_name="Entity",