-
Notifications
You must be signed in to change notification settings - Fork 0
/
types.py
78 lines (68 loc) · 2.25 KB
/
types.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
# © Copyright 2023
# Licensed under the MIT License
# https://opensource.org/licenses/MIT
# https://github.com/OctoDiary
import asyncio
import functools
import inspect
from datetime import datetime, timedelta
from typing import Any, Callable, Union
from pydantic import BaseModel, Field
ReplyMarkup = Union[
dict, list[dict], list[list[dict]],
"ButtonCallback", list["ButtonCallback"], list[list["ButtonCallback"]]
]
class ButtonCallback(BaseModel):
data: str
text: str
callback: Callable[..., Any]
callback_args: Any
callback_kwargs: Any
delete_time: datetime | None
reusable: bool = False
@classmethod
def init(
cls,
data: str,
text: str,
callback: Callable[..., Any],
disable_deadline: bool = False,
reusable: bool = False,
delete_time: datetime | None = None,
*callback_args, **callback_kwargs
) -> "ButtonCallback":
return cls(
data=data,
text=text,
callback=callback,
callback_args=callback_args,
callback_kwargs=callback_kwargs,
delete_time=(
delete_time if delete_time else datetime.now() + timedelta(minutes=20)
) if not disable_deadline else None,
reusable=reusable
)
async def run_callback(self, *args, **kwargs):
if not self.callback:
return None
elif inspect.iscoroutinefunction(self.callback):
return await self.callback(
*(self.callback_args + args),
**{**kwargs, **self.callback_kwargs}
)
else:
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None,
functools.partial(
self.callback,
*(self.callback_args + args),
**{**kwargs, **self.callback_kwargs}
)
)
async def update_callback_args(self, *args, **kwargs):
self.callback_args = args
self.callback_kwargs = kwargs
class AdditionalButtons(BaseModel):
up_buttons: ReplyMarkup = Field([])
below_buttons: ReplyMarkup = Field([])