Use pydantic instead of dataclasses

This commit is contained in:
2024-02-10 12:59:31 +03:00
parent e89cb04c02
commit 3f957db2eb
9 changed files with 70 additions and 187 deletions

View File

@@ -1,8 +1,7 @@
from ..config import Config from pydantic import BaseModel, Field
from ..config import GlobalConfig as Config
class ApiMethod: class ApiMethod(BaseModel):
api_url: str api_url: str = Field(Config().api_url)
def __init__(self) -> None:
self.api_url = Config().api_url

View File

@@ -1,23 +1,6 @@
from dataclasses import dataclass from ._api_model import ApiModel
from typing import Any
from ._helpers import *
@dataclass class Admin(ApiModel):
class Admin:
user_id: int user_id: int
user_name: str user_name: str
@staticmethod
def from_dict(obj: Any) -> 'Admin':
assert isinstance(obj, dict)
user_id = int(from_str(obj.get("user_id")))
user_name = from_str(obj.get("user_name"))
return Admin(user_id, user_name)
def to_dict(self) -> dict:
result: dict = {}
result["user_id"] = from_str(str(self.user_id))
result["user_name"] = from_str(self.user_name)
return result

View File

@@ -0,0 +1,12 @@
from typing import Any, Dict
from pydantic import BaseModel
class ApiModel(BaseModel):
@classmethod
def from_dict(cls: 'ApiModel', obj: Dict[str, Any]) -> 'ApiModel':
return cls(**obj)
def to_dict(self, **kwargs: Any) -> Dict[str, Any]:
return self.model_dump(**kwargs)

View File

@@ -1,8 +1,12 @@
from aiogram import Bot, Dispatcher from aiogram import Bot, Dispatcher
from pydantic import BaseModel
from handlers.handler import Handler from handlers.handler import Handler
class Token(BaseModel):
token: str
class NeuroApiBot: class NeuroApiBot:
bot: Bot bot: Bot
dp: Dispatcher dp: Dispatcher
@@ -10,14 +14,15 @@ class NeuroApiBot:
_instances = {} _instances = {}
def __init__(self, token: str) -> None: def __init__(self, token: str) -> None:
self.bot = Bot(token) token_data = Token(token=token)
self.bot = Bot(token_data.token)
self.dp = Dispatcher() self.dp = Dispatcher()
def __new__(cls, token: str) -> 'NeuroApiBot': def __new__(cls, token: str) -> 'NeuroApiBot':
assert isinstance(token, str) token_data = Token(token=token)
if token not in cls._instances: if token_data.token not in cls._instances:
cls._instances[token] = super(NeuroApiBot, cls).__new__(cls) cls._instances[token_data.token] = super(NeuroApiBot, cls).__new__(cls)
return cls._instances[token] return cls._instances[token_data.token]
def include_router(self, *routerClasses: Handler) -> None: def include_router(self, *routerClasses: Handler) -> None:
for routerClass in routerClasses: for routerClass in routerClasses:

View File

@@ -1,38 +1,17 @@
from dataclasses import dataclass from typing import List
from typing import Optional
from uuid import UUID from uuid import UUID
from ._helpers import * from pydantic import Field
from ._api_model import ApiModel
from ._singleton import Singleton from ._singleton import Singleton
@dataclass class BotSettings(ApiModel, Singleton):
class BotSettings(Singleton):
uuid: UUID uuid: UUID
message_times: List[str] message_times: List[str] = Field([], alias='messageTimes')
channel: str channel: str
is_active: bool is_active: bool = Field(False, alias='isActive')
@staticmethod def get_text(self):
def from_dict(obj: Any) -> 'BotSettings': return f"Канал: {self.channel}\nВремя: {', '.join(self.message_times)}"
assert isinstance(obj, dict)
uuid = UUID(obj.get("uuid"))
message_times = from_list(from_str, obj.get("messageTimes"))
channel = from_str(obj.get("channel"))
is_active = from_bool(obj.get("isActive"))
return BotSettings(uuid, message_times, channel, is_active)
def to_dict(self) -> dict:
result: dict = {}
result["uuid"] = str(self.uuid)
result["messageTimes"] = from_list(from_str, self.message_times)
result["channel"] = from_str(self.channel)
result["isActive"] = from_bool(self.is_active)
return result
@staticmethod
def get_active() -> Optional['BotSettings']:
try:
return BotSettings._instances[BotSettings]
except:
return None

View File

@@ -1,47 +0,0 @@
from datetime import datetime
from typing import Any, Callable, List, TypeVar, Type, cast
import dateutil.parser
T = TypeVar("T")
def from_bool(x: Any) -> bool:
assert isinstance(x, bool)
return x
def from_str(x: Any) -> str:
assert isinstance(x, str)
return x
def from_union(fs, x):
for f in fs:
try:
return f(x)
except:
pass
assert False
def from_none(x: Any) -> Any:
assert x is None
return x
def from_list(f: Callable[[Any], T], x: Any) -> List[T]:
assert isinstance(x, list)
return [f(y) for y in x]
def from_datetime(x: Any) -> datetime:
return dateutil.parser.parse(x)
def to_class(c: Type[T], x: Any) -> dict:
assert isinstance(x, c)
return cast(Any, x).to_dict()
def from_int(x: Any) -> int:
assert isinstance(x, int) and not isinstance(x, bool)
return x

View File

@@ -1,28 +1,10 @@
from dataclasses import dataclass
from uuid import UUID from uuid import UUID
from ._helpers import *
from ._api_model import ApiModel
@dataclass class Image(ApiModel):
class Image:
message_id: int message_id: int
file_id: str file_id: str
has_spoiler: bool has_spoiler: bool
post_uuid: UUID post_uuid: UUID
@staticmethod
def from_dict(obj: Any) -> 'Image':
assert isinstance(obj, dict)
message_id = from_int(obj.get("message_id"))
file_id = from_str(obj.get("file_id"))
has_spoiler = from_bool(obj.get("has_spoiler"))
post_uuid = UUID(obj.get("post_uuid"))
return Image(message_id, file_id, has_spoiler, post_uuid)
def to_dict(self) -> dict:
result: dict = {}
result["message_id"] = from_int(self.message_id)
result["file_id"] = from_str(self.file_id)
result["has_spoiler"] = from_bool(self.has_spoiler)
result["post_uuid"] = str(self.post_uuid)
return result

View File

@@ -1,59 +1,34 @@
import json import json
from dataclasses import dataclass
from datetime import datetime from datetime import datetime
from typing import Any, List, Optional from typing import Any, Dict, List, Optional
from uuid import UUID from uuid import UUID
from aiogram.types import MessageEntity from aiogram.types import MessageEntity
from pydantic import Field
from ._helpers import * from ._api_model import ApiModel
from ._image import Image from ._image import Image
def to_message_dict_class(x: Any) -> dict: class Post(ApiModel):
assert isinstance(x, MessageEntity)
return cast(MessageEntity, x).model_dump()
@dataclass
class Post:
uuid: UUID uuid: UUID
posted: bool posted: bool
text: str text: str
media_group_id: int | str media_group_id: int | str
timestamp: datetime timestamp: datetime
from_user_id: int from_user_id: int
images: Optional[List[Image]] = None images: Optional[List[Image]] = Field(None)
message_entities: Optional[List[MessageEntity]] = None message_entities: Optional[List[MessageEntity]] = Field(None)
@staticmethod
def from_dict(obj: Any) -> 'Post': @classmethod
assert isinstance(obj, dict) def from_dict(cls: 'Post', obj: Dict[str, Any]) -> 'Post':
uuid = UUID(obj.get("uuid"))
posted = from_bool(obj.get("posted"))
text = from_str(obj.get("text"))
media_group_id = from_str(obj.get("media_group_id")) if obj.get(
"media_group_id") is not None else 'None'
timestamp = from_datetime(obj.get("timestamp"))
from_user_id = int(from_str(obj.get("from_user_id")))
images = from_union([lambda x: from_list(
Image.from_dict, x), from_none], obj.get("images"))
mes_ent = json.loads(obj.get('message_entities', '[]')) mes_ent = json.loads(obj.get('message_entities', '[]'))
message_entities = from_list(MessageEntity.model_validate, mes_ent) obj['message_entities'] = mes_ent
return Post(uuid, posted, text, media_group_id, timestamp, from_user_id, images, message_entities) return cls(**obj)
def to_dict(self) -> dict: def to_dict(self) -> dict:
result: dict = {} obj = super().to_dict()
result["uuid"] = str(self.uuid) obj['message_entities'] = json.dumps(obj['message_entities'])
result["posted"] = from_bool(self.posted) obj['media_group_id'] = str(obj['media_group_id'])
result["text"] = from_str(self.text) return obj
result["media_group_id"] = from_str(str(self.media_group_id))
result["timestamp"] = self.timestamp.isoformat()
result["from_user_id"] = from_str(str(self.from_user_id))
if self.images is not None:
result["images"] = from_union([lambda x: from_list(
lambda x: to_class(Image, x), x), from_none], self.images)
if self.message_entities is not None:
result['message_entities'] = from_union([lambda x: from_list(
lambda x: to_message_dict_class(x), x), from_none], self.message_entities)
return result

View File

@@ -1,22 +1,17 @@
from dataclasses import dataclass import json
from typing import Any from typing import Optional
from ._helpers import *
from pydantic import Field
from ._api_model import ApiModel
@dataclass class User(ApiModel):
class User:
id: int id: int
username: str username: str = Field(..., alias='user_name')
banned: Optional[bool] = Field(None)
@staticmethod
def from_dict(obj: Any) -> 'User':
assert isinstance(obj, dict)
id = int(from_str(obj.get("id")))
username = from_str(obj.get("username"))
return User(id, username)
def to_dict(self) -> dict: def to_dict(self) -> dict:
result: dict = {} obj = super().to_dict(exclude_unset=True)
result["id"] = from_str(str(self.id)) obj['id'] = str(obj['id'])
result["username"] = from_str(self.username) return obj
return result