This commit is contained in:
Alexander Kalinovsky
2025-01-04 12:00:12 +01:00
commit 6dbe0536ca
94 changed files with 3467 additions and 0 deletions

46
model/__init__.py Normal file
View File

@@ -0,0 +1,46 @@
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

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

6
model/_singleton.py Normal file
View File

@@ -0,0 +1,6 @@
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]

241
model/bot_entity.py Normal file
View File

@@ -0,0 +1,241 @@
from functools import wraps
from typing import ClassVar, cast, get_args, get_origin
from pydantic import BaseModel
from sqlmodel import SQLModel, BIGINT, Field, select, func
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel.main import SQLModelMetaclass, RelationshipInfo
from .descriptors import EntityDescriptor, EntityField, EntityFieldDescriptor
from .entity_metadata import EntityMetadata
from . import session_dep
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: EntityFieldDescriptor(**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]
field_descriptor = EntityFieldDescriptor(
name = descriptor_name,
field_name = annotation,
type_ = type_,
**descriptor_kwargs)
type_origin = get_origin(type_)
is_list = False
if type_origin == list:
is_list = True
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_ = list[entity_descriptor.type_] if is_list 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()
descriptor_fields_sequence = descriptor_kwargs.pop("field_sequence", None)
if not descriptor_fields_sequence:
descriptor_fields_sequence = list(bot_fields_descriptors.keys())
descriptor_fields_sequence.remove("id")
namespace["bot_entity_descriptor"] = EntityDescriptor(
name = descriptor_name,
class_name = name,
type_ = name,
fields_descriptors = bot_fields_descriptors,
field_sequence = descriptor_fields_sequence,
**descriptor_kwargs)
else:
descriptor_fields_sequence = list(bot_fields_descriptors.keys())
descriptor_fields_sequence.remove("id")
descriptor_name = name.lower()
namespace["bot_entity_descriptor"] = EntityDescriptor(
name = descriptor_name,
class_name = name,
type_ = name,
fields_descriptors = bot_fields_descriptors,
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"] == True:
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]:
field_descriptor.type_ = list[type_] if get_origin(field_descriptor.type_) == list else type_
a = field_descriptor
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 = Field(
primary_key = True,
sa_type = BIGINT)
name: str
@classmethod
@session_dep
async def get(cls, *,
session: AsyncSession | None = None,
id: int):
return await session.get(cls, id)
@classmethod
@session_dep
async def get_count(cls, *,
session: AsyncSession | None = None) -> int:
return await session.scalar(select(func.count()).select_from(cls))
@classmethod
@session_dep
async def get_multi(cls, *,
session: AsyncSession | None = None,
order_by = None,
skip: int = 0,
limit: int = None):
select_statement = select(cls).offset(skip)
if limit:
select_statement = select_statement.limit(limit)
if order_by:
select_statement = select_statement.order_by(order_by)
return (await session.exec(select_statement)).all()
@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

147
model/bot_enum.py Normal file
View File

@@ -0,0 +1,147 @@
from pydantic_core.core_schema import str_schema
from sqlalchemy.types import TypeDecorator, String
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 owner.all_members.items()}[self.value]
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) -> bool:
if other is None:
return False
if isinstance(other, str):
return self.value == other
return self.value == other.value
def __hash__(self):
return hash(self.value)
def localized(self, lang: str = None) -> str:
if self.loc_obj and len(self.loc_obj) > 0:
if lang and 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 = String(256)
def __init__(self, enum_type: BotEnum):
self._enum_type = enum_type
super().__init__()
def process_bind_param(self, value, dialect):
if value and isinstance(value, EnumMember):
return value.value
return None
def process_result_value(self, value, dialect):
if value:
return self._enum_type(value)
return None

4
model/default_user.py Normal file
View File

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

90
model/descriptors.py Normal file
View File

@@ -0,0 +1,90 @@
from typing import Any, Callable
from babel.support import LazyProxy
from dataclasses import dataclass, field
from .role import RoleBase
from . import EntityPermission
EntityCaptionCallable = Callable[["EntityDescriptor"], str]
EntityItemCaptionCallable = Callable[["EntityDescriptor", Any], str]
EntityFieldCaptionCallable = Callable[["EntityFieldDescriptor", Any, Any], str]
@dataclass(kw_only = True)
class _BaseEntityFieldDescriptor():
icon: str = None
caption_str: str | LazyProxy | EntityFieldCaptionCallable | None = None
caption_btn: str | LazyProxy | EntityFieldCaptionCallable | None = None
description: str | LazyProxy | EntityFieldCaptionCallable | None = None
edit_prompt: str | LazyProxy | EntityFieldCaptionCallable | None = None
caption_value_str: str | LazyProxy | EntityFieldCaptionCallable | None = None
caption_value_btn: str | LazyProxy | EntityFieldCaptionCallable | None = None
is_visible: bool = True
localizable: bool = False
bool_false_value: str | LazyProxy = "no"
bool_false_value_btn: str | LazyProxy = "no"
bool_true_value: str | LazyProxy = "yes"
bool_true_value_btn: str | LazyProxy = "yes"
default: Any = None
@dataclass(kw_only = True)
class EntityField(_BaseEntityFieldDescriptor):
name: str | None = None
sm_descriptor: Any = None
@dataclass(kw_only = True)
class Setting(_BaseEntityFieldDescriptor):
name: str | None = None
@dataclass(kw_only = True)
class EntityFieldDescriptor(_BaseEntityFieldDescriptor):
name: str
field_name: str
type_: type
entity_descriptor: "EntityDescriptor" = None
def __hash__(self):
return self.name.__hash__()
@dataclass(kw_only = True)
class _BaseEntityDescriptor:
icon: str = "📘"
caption_msg: str | LazyProxy | EntityCaptionCallable | None = None
caption_btn: str | LazyProxy | EntityCaptionCallable | None = None
description: str | LazyProxy | EntityCaptionCallable | None = None
item_caption_msg: EntityItemCaptionCallable | None = None
item_caption_btn: EntityItemCaptionCallable | None = None
show_in_entities_menu: bool = True
field_sequence: list[str] = None
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
fields_descriptors: dict[str, EntityFieldDescriptor]

8
model/entity_metadata.py Normal file
View File

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

45
model/field_types.py Normal file
View File

@@ -0,0 +1,45 @@
from dataclasses import dataclass
from babel.support import LazyProxy
from typing import TypeVar
from .bot_entity import BotEntity
@dataclass
class FieldType: ...
class LocStr(str): ...
class String(FieldType):
localizable: bool = False
class Integer(FieldType):
pass
@dataclass
class Decimal:
precision: int = 0
@dataclass
class Boolean:
true_value: str | LazyProxy = "true"
false_value: str | LazyProxy = "false"
@dataclass
class DateTime:
pass
EntityType = TypeVar('EntityType', bound = BotEntity)
@dataclass
class EntityReference:
entity_type: type[EntityType]

8
model/fsm_storage.py Normal file
View File

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

6
model/language.py Normal file
View File

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

44
model/owned_bot_entity.py Normal file
View File

@@ -0,0 +1,44 @@
from sqlmodel import BIGINT, Field, select, func
from sqlmodel.ext.asyncio.session import AsyncSession
from .bot_entity import BotEntity
from .descriptors import EntityField
from .user import UserBase
from . import session_dep
class OwnedBotEntity(BotEntity, table = False):
user_id: int | None = EntityField(
sm_descriptor = Field(sa_type = BIGINT, foreign_key = "user.id", ondelete="SET NULL"),
is_visible = False)
@classmethod
@session_dep
async def get_multi_by_user(cls, *,
session: AsyncSession | None = None,
user_id: int,
order_by = None,
skip: int = 0,
limit: int = None):
select_statement = select(cls).where(cls.user_id == user_id).offset(skip)
if limit:
select_statement = select_statement.limit(limit)
if order_by:
select_statement = select_statement.order_by(order_by)
return (await session.exec(select_statement)).all()
@classmethod
@session_dep
async def get_count_by_user(cls, *,
session: AsyncSession | None = None,
user_id: int):
return await session.scalar(
select(func.count()).
select_from(cls).
where(cls.user_id == user_id))

7
model/role.py Normal file
View File

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

183
model/settings.py Normal file
View File

@@ -0,0 +1,183 @@
from datetime import datetime
from sqlmodel import SQLModel, Field, select
from typing import Any, get_origin
from ..db import async_session
from .role import RoleBase
from .descriptors import EntityFieldDescriptor, Setting
from ..utils import deserialize, serialize
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['__annotations__']:
if annotation in ["_settings_descriptors", "_cache", "_cached_settings"]:
continue
attr_value = attributes.get(annotation)
name = annotation
if isinstance(attr_value, Setting):
descriptor_kwargs = attr_value.__dict__.copy()
name = descriptor_kwargs.pop("name") or annotation
attributes[annotation] = EntityFieldDescriptor(
name = name,
field_name = annotation,
type_ = attributes['__annotations__'][annotation],
**descriptor_kwargs)
else:
attributes[annotation] = EntityFieldDescriptor(
name = annotation,
field_name = annotation,
type_ = attributes['__annotations__'][annotation],
default = attr_value)
settings_descriptors[name] = 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, EntityFieldDescriptor] = {}
PAGE_SIZE: int = Setting(default = 10, )
SECURITY_SETTINGS_ROLES: list[RoleBase] = [RoleBase.SUPER_USER]
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_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_INVALID_INPUT: str = Setting(name = "AS_INVALID_INPUT", default = "Invalid input", is_visible = False)
@classmethod
async def get[T](cls, param: T) -> T:
name = param.field_name
if param.name not in cls._cache.keys():
cls._cache[name] = await cls.load_param(param)
return cls._cache[name]
@classmethod
async def load_param(cls, param: EntityFieldDescriptor) -> 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_ == 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]
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 | EntityFieldDescriptor, 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, EntityFieldDescriptor]:
return cls._settings_descriptors
@classmethod
async def get_params(cls) -> dict[EntityFieldDescriptor, Any]:
params = cls.list_params()
return {param: await cls.get(param) for _, param in params.items()}

19
model/user.py Normal file
View File

@@ -0,0 +1,19 @@
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
class UserBase(BotEntity, table = False):
__tablename__ = "user"
lang: LanguageBase = Field(sa_type = EnumType(LanguageBase), default = LanguageBase.DEFAULT)
is_active: bool = True
roles: list[RoleBase] = Field(sa_type=ARRAY(EnumType(RoleBase)), default = [RoleBase.DEFAULT_USER])