Перейти к содержанию

MaxApi

MaxApi — асинхронный высокоуровневый клиент. Он владеет выбранным backend-стеком и предоставляет операции для сообщений, чатов, реакций, опросов, групп, контактов, профиля, папок, сессий, presence и двухфакторной авторизации.

Bases: AsyncInitializerMixin, FullMixin

Asynchronous client for MAX Messenger.

The client initializes a transport, protocol, and mapper from the project registry. Initialization is asynchronous and requires the selected backend names to be available in the corresponding registries.

Raises:

Type Description
RuntimeError

If a transport, protocol, or mapper name is not supported.

Initialize the max api.

Parameters:

Name Type Description Default
device_type str

Device type reported to the API.

'WEB'
password str | None

Optional account password.

None
token str | None

Optional auth token.

None
transport BaseTransport | None

Transport backend name.

None
protocol BaseMaxProtocol[Any, Any] | None

Protocol backend name.

None
mapper BaseMapper[Any, Any] | None

Mapper backend name.

None
transport_options dict[str, Any] | None

Keyword arguments passed to the transport constructor.

None
kwargs Any

Extra keyword arguments passed to mapper initialization.

{}
workflow_data dict[Any, Any] | None

dict[Any, Any] global workflow data.

None
user_agent_params dict[str, Any] | None

dict[str, Any] params of user agent.

required
auth_middleware_manager AuthMiddlewareManager | None

AuthMiddlewareManager instance of auth middleware manager.

None
registration_config RegistrationConfig | None

instance of RegistrationConfig for register account.

None
token_suffix str | None

The token suffix value.

None

Raises:

Type Description
RuntimeError

If transport or protocol or mapper cannot be None.

Source code in src/pyromax/core/client.py
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
def __init__(
    self,
    device_type: str = "WEB",
    password: str | None = None,
    transport: BaseTransport | None = None,
    protocol: BaseMaxProtocol[Any, Any] | None = None,
    mapper: BaseMapper[Any, Any] | None = None,
    transport_options: dict[str, Any] | None = None,
    token: str | None = None,
    logger: logging.Logger | None = None,
    workflow_data: dict[Any, Any] | None = None,
    auth_middleware_manager: AuthMiddlewareManager | None = None,
    registration_config: RegistrationConfig | None = None,
    token_suffix: str | None = None,
    **kwargs: Any,
) -> None:
    """Initialize the max api.

    :param device_type: Device type reported to the API.
    :type device_type: str
    :param password: Optional account password.
    :type password: str | None
    :param token: Optional auth token.
    :type token: str | None
    :param transport: Transport backend name.
    :type transport: str
    :param protocol: Protocol backend name.
    :type protocol: str
    :param mapper: Mapper backend name.
    :type mapper: str
    :param transport_options: Keyword arguments passed to the transport constructor.
    :type transport_options: dict[str, Any] | None
    :param kwargs: Extra keyword arguments passed to mapper initialization.
    :type kwargs: Any

    :param workflow_data: dict[Any, Any] global workflow data.
    :type workflow_data: dict[Any, Any] | None
    :param user_agent_params: dict[str, Any] params of user agent.
    :type user_agent_params: dict[str, Any] | None
    :param auth_middleware_manager: AuthMiddlewareManager instance of auth middleware manager.
    :type auth_middleware_manager: AuthMiddlewareManager | None
    :param registration_config: instance of RegistrationConfig for register account.
    :type registration_config: RegistrationConfig | None
    :param token_suffix: The token suffix value.
    :type token_suffix: str | None
    :raises RuntimeError: If transport or protocol or mapper cannot be None.
    """
    if workflow_data is None:
        workflow_data = {}

    if logger is None:
        logger = logging.getLogger("MaxApi")

    if transport is None or protocol is None or mapper is None:
        raise RuntimeError("transport or protocol or mapper cannot be None")

    self.transport = transport
    self.transport_options = transport_options
    self.protocol = protocol
    self.mapper = mapper
    self.token = token
    self.password = password
    self.id: int | None = None
    self.phone: str | None = None

    self.me: Profile | None = None
    self.chats: list[Chat] | None = None
    self.names: list[Name] | None = None
    self.contacts: list[Contact | None] = []
    self.users: dict[int, Contact] = {}

    self._logger: logging.Logger | None = logger
    self.workflow_data = workflow_data
    self.auth_middleware_manager = auth_middleware_manager

transport instance-attribute

transport = transport

transport_options instance-attribute

transport_options = transport_options

protocol instance-attribute

protocol = protocol

mapper instance-attribute

mapper = mapper

token instance-attribute

token = token

password instance-attribute

password = password

id instance-attribute

id: int | None = None

phone instance-attribute

phone: str | None = None

me instance-attribute

me: Profile | None = None

chats instance-attribute

chats: list[Chat] | None = None

names instance-attribute

names: list[Name] | None = None

contacts instance-attribute

contacts: list[Contact | None] = []

users instance-attribute

users: dict[int, Contact] = {}

_logger instance-attribute

_logger: Logger | None = logger

workflow_data instance-attribute

workflow_data = workflow_data

auth_middleware_manager instance-attribute

auth_middleware_manager = auth_middleware_manager

_async_init async

_async_init(
    device_type: str = "WEB",
    password: str | None = None,
    token: str | None = None,
    transport: str = "websocket",
    protocol: str = "EnvelopeProtocol",
    mapper: str = "EnvelopeV11",
    transport_options: dict[str, Any] | None = None,
    workflow_data: dict[Any, Any] | None = None,
    user_agent_params: dict[str, Any] | None = None,
    auth_middleware_manager: (
        AuthMiddlewareManager | None
    ) = None,
    registration_config: RegistrationConfig | None = None,
    token_suffix: str | None = None,
    **kwargs: Any
) -> None

Asynchronously initialize transport, protocol, and mapper.

Parameters:

Name Type Description Default
device_type str

Device type reported to the API.

'WEB'
password str | None

Optional account password.

None
token str | None

Optional auth token.

None
transport str

Transport backend name.

'websocket'
protocol str

Protocol backend name.

'EnvelopeProtocol'
mapper str

Mapper backend name.

'EnvelopeV11'
transport_options dict[str, Any] | None

Keyword arguments passed to the transport constructor.

None
kwargs Any

Extra keyword arguments passed to mapper initialization.

{}
workflow_data dict[Any, Any] | None

dict[Any, Any] global workflow data.

None
user_agent_params dict[str, Any] | None

dict[str, Any] params of user agent.

None
auth_middleware_manager AuthMiddlewareManager | None

AuthMiddlewareManager instance of auth middleware manager.

None
registration_config RegistrationConfig | None

instance of RegistrationConfig for register account.

None
token_suffix str | None

The token suffix value.

None

Raises:

Type Description
RuntimeError

If transport or protocol or mapper cannot be None.

Source code in src/pyromax/core/client.py
 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
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
async def _async_init(
    self,
    device_type: str = "WEB",
    password: str | None = None,
    token: str | None = None,
    transport: str = "websocket",
    protocol: str = "EnvelopeProtocol",
    mapper: str = "EnvelopeV11",
    transport_options: dict[str, Any] | None = None,
    workflow_data: dict[Any, Any] | None = None,
    user_agent_params: dict[str, Any] | None = None,
    auth_middleware_manager: AuthMiddlewareManager | None = None,
    registration_config: RegistrationConfig | None = None,
    token_suffix: str | None = None,
    **kwargs: Any,
) -> None:
    """Asynchronously initialize transport, protocol, and mapper.

    :param device_type: Device type reported to the API.
    :type device_type: str
    :param password: Optional account password.
    :type password: str | None
    :param token: Optional auth token.
    :type token: str | None
    :param transport: Transport backend name.
    :type transport: str
    :param protocol: Protocol backend name.
    :type protocol: str
    :param mapper: Mapper backend name.
    :type mapper: str
    :param transport_options: Keyword arguments passed to the transport constructor.
    :type transport_options: dict[str, Any] | None
    :param kwargs: Extra keyword arguments passed to mapper initialization.
    :type kwargs: Any

    :param workflow_data: dict[Any, Any] global workflow data.
    :type workflow_data: dict[Any, Any] | None
    :param user_agent_params: dict[str, Any] params of user agent.
    :type user_agent_params: dict[str, Any] | None
    :param auth_middleware_manager: AuthMiddlewareManager instance of auth middleware manager.
    :type auth_middleware_manager: AuthMiddlewareManager | None
    :param registration_config: instance of RegistrationConfig for register account.
    :type registration_config: RegistrationConfig | None
    :param token_suffix: The token suffix value.
    :type token_suffix: str | None
    :raises RuntimeError: If transport or protocol or mapper cannot be None.
    """
    if workflow_data is None:
        workflow_data = {}

    logger = logging.getLogger("MaxApi")

    if transport not in TRANSPORTS:
        raise RuntimeError(f"transport {transport} is not supported")

    if protocol not in PROTOCOLS:
        raise RuntimeError(f"protocol {protocol} is not supported")

    if mapper not in MAPPERS:
        raise RuntimeError(f"mapper {mapper} is not supported")

    logger.info("Start initialization...")

    logger.info("Initializing transport...")
    if transport_options:
        max_transport = await TRANSPORTS[transport](**transport_options)
    else:
        max_transport = await TRANSPORTS[transport]()
    logger.info("Transport initialized.")

    logger.info("Initializing protocol...")
    protocol_res: Any = await PROTOCOLS[protocol](transport=max_transport)
    max_protocol: BaseMaxProtocol[Any, Any] = protocol_res
    logger.info("Protocol initialized.")

    logger.info("Initializing mapper...")
    map_class = MAPPERS[mapper]
    max_mapper = await map_class(self, protocol=max_protocol)
    logger.info("Mapper initialized.")

    await asyncio.to_thread(
        self.__init__,  # type: ignore[misc]
        protocol=max_protocol,
        password=password,
        transport=max_transport,
        mapper=max_mapper,
        transport_options=transport_options,
        token=token,
        logger=logger,
        workflow_data=workflow_data,
        device_type=device_type,
        auth_middleware_manager=auth_middleware_manager,
    )

    if token is None and self.auth_middleware_manager is not None:
        from ..models.AuthFlow import AuthFlow

        mapper_type = type(self.mapper)
        protocol_type = type(self.protocol)
        transport_type = type(self.transport)

        auth_alias = AuthFlow[
            mapper_type,  # type: ignore[valid-type]
            protocol_type,  # type: ignore[valid-type]
            transport_type,  # type: ignore[valid-type]
        ]

        async def auth_wrapped(
            auth_flow: AuthFlow[Any, Any, Any],
            _: dict[Any, Any],
        ) -> AuthFlow[Any, Any, Any]:
            """Auth wrapped.

            :param auth_flow: AuthFlow[Any, Any, Any] instance to process.
            :type auth_flow: AuthFlow[Any, Any, Any]
            :param _: dict[Any, Any] instance to process.
            :type _: dict[Any, Any]
            :returns: The resulting AuthFlow[Any, Any, Any] value.
            :rtype: AuthFlow[Any, Any, Any]
            """
            return auth_flow

        wrapped = self.auth_middleware_manager.wrap_middlewares(
            self.auth_middleware_manager,
            auth_wrapped,
        )

        auth_alias.model_rebuild(
            _types_namespace={
                "MaxApi": type(self),
            }
        )

        flow = auth_alias(
            mapper=self.mapper,
            protocol=self.protocol,
            transport=self.transport,
        )
        flow.as_(self)

        data = {
            type(self): self,
            mapper_type: self.mapper,
            protocol_type: self.protocol,
            transport_type: self.transport,
        }

        resolved_flow = await wrapped(flow, cast(dict[Any, Any], data))
        token = resolved_flow.token

    await self.mapper.initialize_client(
        token=token,
        device_type=device_type,
        password=password,
        user_agent_params=user_agent_params,
        registration_config=registration_config,
        token_suffix=token_suffix,
        **kwargs,
    )

__call__ async

__call__(
    class_of_method: type[BaseMaxApiMethod[Any]],
    *args: Any,
    **kwargs: Any
) -> Any

Invoke the max api.

Parameters:

Name Type Description Default
class_of_method type[BaseMaxApiMethod[Any]]

MAX API method class to instantiate and execute.

required
args Any

Positional arguments forwarded to the wrapped callable.

()
kwargs Any

Keyword arguments forwarded to the wrapped callable.

{}

Returns:

Type Description
Any

The value returned by the wrapped callable or backend.

Raises:

Type Description
RuntimeError

If try a call method before initialization, because logger has not been initialized.

Source code in src/pyromax/core/client.py
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
async def __call__(
    self, class_of_method: type[BaseMaxApiMethod[Any]], *args: Any, **kwargs: Any
) -> Any:
    """Invoke the max api.

    :param class_of_method: MAX API method class to instantiate and execute.
    :type class_of_method: type[BaseMaxApiMethod[Any]]
    :param args: Positional arguments forwarded to the wrapped callable.
    :type args: Any
    :param kwargs: Keyword arguments forwarded to the wrapped callable.
    :type kwargs: Any
    :returns: The value returned by the wrapped callable or backend.
    :rtype: Any
    :raises RuntimeError: If try a call method before initialization, because logger has not been initialized.
    """
    if self._logger is None:
        raise RuntimeError(
            "Try a call method before initialization, because logger has not been initialized"
        )
    self._logger.debug("Calling MaxApi method: %s", class_of_method.__name__)
    method = class_of_method().as_(self)

    return await method(*args, **kwargs)

listen_updates

listen_updates(
    context: Any,
) -> tuple[
    Callable[[Response], MaxObject],
    AsyncGenerator[Response, None],
]

Yield incoming updates forever.

Parameters:

Name Type Description Default
context Any

Runtime context passed to the mapper.

required

Returns:

Type Description
tuple[Callable[[Response], MaxObject], AsyncGenerator[Response, None]]

Stream of incoming updates.

Source code in src/pyromax/core/client.py
300
301
302
303
304
305
306
307
308
309
310
311
def listen_updates(
    self, context: Any
) -> tuple[Callable[[Response], MaxObject], AsyncGenerator[Response, None]]:
    """Yield incoming updates forever.

    :param context: Runtime context passed to the mapper.
    :type context: Any

    :returns: Stream of incoming updates.
    :rtype: tuple[Callable[[Response], MaxObject], AsyncGenerator[Response, None]]
    """
    return self.mapper.listen_updates(context=context)

get_members_by_ids async

get_members_by_ids(
    member_ids: list[int],
) -> Sequence[Contact]

Retrieve members by ids.

Parameters:

Name Type Description Default
member_ids list[int]

Identifiers of the members.

required

Returns:

Type Description
Sequence[Contact]

The Contacts collection.

Source code in src/pyromax/core/CoreMixins/Contacts.py
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
async def get_members_by_ids(self, member_ids: list[int]) -> Sequence[Contact]:
    """Retrieve members by ids.

    :param member_ids: Identifiers of the members.
    :type member_ids: list[int]
    :returns: The Contacts collection.
    :rtype: Sequence[Contact]
    """

    contacts = cast(
        Sequence[Contact],
        await self(
            GetMembersByIdsMethod,
            member_ids=member_ids,
        ),
    )
    return contacts

get_member_by_id async

get_member_by_id(member_id: int) -> Contact | None

Retrieve member by id.

Parameters:

Name Type Description Default
member_id int

Identifier of the member.

required

Returns:

Type Description
Contact | None

The resulting Contact | None.

Source code in src/pyromax/core/CoreMixins/Contacts.py
42
43
44
45
46
47
48
49
50
51
async def get_member_by_id(self, member_id: int) -> Contact | None:
    """Retrieve member by id.

    :param member_id: Identifier of the member.
    :type member_id: int
    :returns: The resulting Contact | None.
    :rtype: Contact | None
    """
    contacts = await self.get_members_by_ids(member_ids=[member_id])
    return contacts[0] if contacts else None

get_users async

get_users(user_ids: list[int]) -> list[Contact]

Retrieve users.

Parameters:

Name Type Description Default
user_ids list[int]

Identifiers of the users.

required

Returns:

Type Description
list[Contact]

The resulting collection.

Source code in src/pyromax/core/CoreMixins/Contacts.py
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
async def get_users(self, user_ids: list[int]) -> list[Contact]:
    """Retrieve users.

    :param user_ids: Identifiers of the users.
    :type user_ids: list[int]
    :returns: The resulting collection.
    :rtype: list[Contact]
    """

    user = cast(
        list[Contact],
        await self(
            GetUsersMethod,
            user_ids=user_ids,
        ),
    )
    return user

get_user async

get_user(user_id: int) -> Contact | None

Retrieve user.

Parameters:

Name Type Description Default
user_id int

Identifier of the user.

required

Returns:

Type Description
Contact | None

The resulting Contact | None.

Source code in src/pyromax/core/CoreMixins/Contacts.py
71
72
73
74
75
76
77
78
79
80
async def get_user(self, user_id: int) -> Contact | None:
    """Retrieve user.

    :param user_id: Identifier of the user.
    :type user_id: int
    :returns: The resulting Contact | None.
    :rtype: Contact | None
    """
    user = await self.get_users(user_ids=[user_id])
    return user[0] if user else None

search_by_phone async

search_by_phone(phone: str) -> Contact

Search for by phone.

Parameters:

Name Type Description Default
phone str

Phone number in the format accepted by MAX.

required

Returns:

Type Description
Contact

The resulting Contact.

Source code in src/pyromax/core/CoreMixins/Contacts.py
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
async def search_by_phone(self, phone: str) -> Contact:
    """Search for by phone.

    :param phone: Phone number in the format accepted by MAX.
    :type phone: str
    :returns: The resulting Contact.
    :rtype: Contact
    """

    user = cast(
        Contact,
        await self(
            SearchByPhoneMethod,
            phone=phone,
        ),
    )
    return user

get_sessions async

get_sessions() -> list[Session]

Retrieve sessions.

Returns:

Type Description
list[Session]

The Sessions collection.

Source code in src/pyromax/core/CoreMixins/Contacts.py
100
101
102
103
104
105
106
107
async def get_sessions(self) -> list[Session]:
    """Retrieve sessions.

    :returns: The Sessions collection.
    :rtype: list[Session]
    """

    return cast(list[Session], await self(GetSessionsMethod))

get_chat_id async

get_chat_id(first_user_id: int, second_user_id: int) -> int

Retrieve chat id.

Parameters:

Name Type Description Default
first_user_id int

Identifier of the first user.

required
second_user_id int

Identifier of the second user.

required

Returns:

Type Description
int

The resulting int value.

Source code in src/pyromax/core/CoreMixins/Contacts.py
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
async def get_chat_id(self, first_user_id: int, second_user_id: int) -> int:
    """Retrieve chat id.

    :param first_user_id: Identifier of the first user.
    :type first_user_id: int
    :param second_user_id: Identifier of the second user.
    :type second_user_id: int
    :returns: The resulting int value.
    :rtype: int
    """
    return cast(
        int,
        await self(
            GetChatIdMethod,
            first_user_id=first_user_id,
            second_user_id=second_user_id,
        ),
    )

add_contact async

add_contact(contact_id: int) -> Contact

Add contact.

Parameters:

Name Type Description Default
contact_id int

Identifier of the contact.

required

Returns:

Type Description
Contact

The resulting Contact.

Source code in src/pyromax/core/CoreMixins/Contacts.py
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
async def add_contact(self, contact_id: int) -> Contact:
    """Add contact.

    :param contact_id: Identifier of the contact.
    :type contact_id: int
    :returns: The resulting Contact.
    :rtype: Contact
    """

    return cast(
        Contact,
        await self(
            AddContactMethod,
            contact_id=contact_id,
        ),
    )

remove_contact async

remove_contact(contact_id: int) -> None

Remove contact.

Parameters:

Name Type Description Default
contact_id int

Identifier of the contact.

required
Source code in src/pyromax/core/CoreMixins/Contacts.py
145
146
147
148
149
150
151
152
153
154
155
156
157
async def remove_contact(self, contact_id: int) -> None:
    """Remove contact.

    :param contact_id: Identifier of the contact.
    :type contact_id: int
    """
    return cast(
        None,
        await self(
            RemoveContactMethod,
            contact_id=contact_id,
        ),
    )

import_contacts async

import_contacts(
    contacts: list[ContactInfo],
) -> list[Contact]

Import contacts.

Parameters:

Name Type Description Default
contacts list[ContactInfo]

Collection of contacts.

required

Returns:

Type Description
list[Contact]

The Contacts collection.

Source code in src/pyromax/core/CoreMixins/Contacts.py
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
async def import_contacts(self, contacts: list[ContactInfo]) -> list[Contact]:
    """Import contacts.

    :param contacts: Collection of contacts.
    :type contacts: list[ContactInfo]
    :returns: The Contacts collection.
    :rtype: list[Contact]
    """

    return cast(
        list[Contact],
        await self(
            ImportContactsMethod,
            contacts=contacts,
        ),
    )

download_file async

download_file(
    file: BaseFileAttachment,
) -> tuple[bytes, dict[str, str]] | tuple[None, None]

Download file.

Parameters:

Name Type Description Default
file BaseFileAttachment

File attachment to process.

required

Returns:

Type Description
tuple[bytes, dict[str, str]] | tuple[None, None]

The resulting tuple[bytes, dict[str, str]] | tuple[None, None] value is request headers | None semantic.

Source code in src/pyromax/core/CoreMixins/File.py
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
async def download_file(
    self, file: BaseFileAttachment
) -> tuple[bytes, dict[str, str]] | tuple[None, None]:
    """Download file.

    :param file: File attachment to process.
    :type file: BaseFileAttachment
    :returns: The resulting tuple[bytes, dict[str, str]] | tuple[None, None] value is request headers | None semantic.
    :rtype: tuple[bytes, dict[str, str]] | tuple[None, None]
    """
    return cast(
        tuple[bytes, dict[str, str]] | tuple[None, None],
        await self(
            DownloadFileMethod,
            file=file,
        ),
    )

upload_file async

upload_file(
    data: bytes | None,
    typeof: type[BaseFileAttachment],
    **kwargs: Any
) -> list[BaseFileAttachment | Any]

Upload file.

Parameters:

Name Type Description Default
data bytes | None

Contextual data passed through the processing pipeline.

required
typeof type[BaseFileAttachment]

Attachment class that determines the upload type.

required
kwargs Any

Keyword arguments forwarded to the wrapped callable.

{}

Returns:

Type Description
list[BaseFileAttachment | Any]

The resulting collection.

Source code in src/pyromax/core/CoreMixins/File.py
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
async def upload_file(
    self, data: bytes | None, typeof: type[BaseFileAttachment], **kwargs: Any
) -> list[BaseFileAttachment | Any]:
    """Upload file.

    :param data: Contextual data passed through the processing pipeline.
    :type data: bytes | None
    :param typeof: Attachment class that determines the upload type.
    :type typeof: type[BaseFileAttachment]
    :param kwargs: Keyword arguments forwarded to the wrapped callable.
    :type kwargs: Any
    :returns: The resulting collection.
    :rtype: list[BaseFileAttachment | Any]
    """
    # from ..models import BaseFileAttachment

    return cast(
        list[BaseFileAttachment | Any],
        await self(
            UploadFileMethod,
            data=data,
            typeof=typeof,
            **kwargs,
        ),
    )

change_profile async

change_profile(
    first_name: str,
    last_name: str | None = None,
    description: str | None = None,
    photo: bytes | None = None,
    file_name: str | None = None,
    photo_token: str | None = None,
) -> Profile

Change profile.

Parameters:

Name Type Description Default
first_name str

The first name.

required
last_name str | None

The last name.

None
description str | None

The description.

None
photo bytes | None

The photo data.

None
file_name str | None

The file name or photo file.

None
photo_token str | None

The photo token value.

None

Returns:

Type Description
Profile

The resulting Profile.

Source code in src/pyromax/core/CoreMixins/User.py
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
async def change_profile(
    self,
    first_name: str,
    last_name: str | None = None,
    description: str | None = None,
    photo: bytes | None = None,
    file_name: str | None = None,
    photo_token: str | None = None,
) -> Profile:
    """Change profile.

    :param first_name: The first name.
    :type first_name: str
    :param last_name: The last name.
    :type last_name: str | None
    :param description: The description.
    :type description: str | None
    :param photo: The photo data.
    :type photo: bytes | None
    :param file_name: The file name or photo file.
    :type file_name: str | None
    :param photo_token: The photo token value.
    :type photo_token: str | None
    :returns: The resulting Profile.
    :rtype: Profile
    """

    return cast(
        Profile,
        await self(
            ChangeProfileMethod,
            first_name=first_name,
            last_name=last_name,
            description=description,
            photo=photo,
            file_name=file_name,
            photo_token=photo_token,
        ),
    )

create_folder async

create_folder(
    title: str,
    chat_include: list[int],
    filters: list[Any] | None = None,
    folder_id: str | None = None,
) -> FolderUpdate

Create folder.

Parameters:

Name Type Description Default
title str

The title of folder.

required
chat_include list[int]

Collection of chat include.

required
filters list[Any] | None

Collection of filters.

None
folder_id str | None

Identifier of the folder.

None

Returns:

Type Description
FolderUpdate

The resulting FolderUpdate.

Source code in src/pyromax/core/CoreMixins/User.py
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
async def create_folder(
    self,
    title: str,
    chat_include: list[int],
    filters: list[Any] | None = None,
    folder_id: str | None = None,
) -> FolderUpdate:
    """Create folder.

    :param title: The title of folder.
    :type title: str
    :param chat_include: Collection of chat include.
    :type chat_include: list[int]
    :param filters: Collection of filters.
    :type filters: list[Any] | None
    :param folder_id: Identifier of the folder.
    :type folder_id: str | None
    :returns: The resulting FolderUpdate.
    :rtype: FolderUpdate
    """

    return cast(
        FolderUpdate,
        await self(
            CreateFolderMethod,
            title=title,
            chat_include=chat_include,
            filters=filters,
            folder_id=folder_id,
        ),
    )

get_folders async

get_folders(folder_sync: int = 0) -> FolderList

Retrieve folders.

Parameters:

Name Type Description Default
folder_sync int

Synchronization marker. Leave as 0 for the initial load..

0

Returns:

Type Description
FolderList

The resulting FolderList.

Source code in src/pyromax/core/CoreMixins/User.py
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
async def get_folders(self, folder_sync: int = 0) -> FolderList:
    """Retrieve folders.

    :param folder_sync: Synchronization marker. Leave as ``0`` for the initial load..
    :type folder_sync: int
    :returns: The resulting FolderList.
    :rtype: FolderList
    """

    return cast(
        FolderList,
        await self(
            GetFoldersMethod,
            folder_sync=folder_sync,
        ),
    )

update_folder async

update_folder(
    folder_id: str,
    title: str,
    chat_include: list[int] | None = None,
    filters: list[Any] | None = None,
    options: list[Any] | None = None,
) -> FolderUpdate

Update folder.

Parameters:

Name Type Description Default
folder_id str

Identifier of the folder.

required
title str

The title of folder.

required
chat_include list[int] | None

Collection of chat include.

None
filters list[Any] | None

Collection of filters.

None
options list[Any] | None

Collection of options.

None

Returns:

Type Description
FolderUpdate

The resulting FolderUpdate.

Source code in src/pyromax/core/CoreMixins/User.py
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
async def update_folder(
    self,
    folder_id: str,
    title: str,
    chat_include: list[int] | None = None,
    filters: list[Any] | None = None,
    options: list[Any] | None = None,
) -> FolderUpdate:
    """Update folder.

    :param folder_id: Identifier of the folder.
    :type folder_id: str
    :param title: The title of folder.
    :type title: str
    :param chat_include: Collection of chat include.
    :type chat_include: list[int] | None
    :param filters: Collection of filters.
    :type filters: list[Any] | None
    :param options: Collection of options.
    :type options: list[Any] | None
    :returns: The resulting FolderUpdate.
    :rtype: FolderUpdate
    """

    return cast(
        FolderUpdate,
        await self(
            UpdateFolderMethod,
            title=title,
            chat_include=chat_include,
            filters=filters,
            folder_id=folder_id,
            options=options,
        ),
    )

delete_folders async

delete_folders(folder_ids: list[str]) -> FolderUpdate

Delete folders.

Parameters:

Name Type Description Default
folder_ids list[str]

Identifiers of the folders.

required

Returns:

Type Description
FolderUpdate

The resulting FolderUpdate.

Source code in src/pyromax/core/CoreMixins/User.py
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
async def delete_folders(
    self,
    folder_ids: list[str],
) -> FolderUpdate:
    """Delete folders.

    :param folder_ids: Identifiers of the folders.
    :type folder_ids: list[str]
    :returns: The resulting FolderUpdate.
    :rtype: FolderUpdate
    """

    return cast(
        FolderUpdate,
        await self(
            DeleteFoldersMethod,
            folder_ids=folder_ids,
        ),
    )

delete_folder async

delete_folder(folder_id: str) -> FolderUpdate

Delete folder.

Parameters:

Name Type Description Default
folder_id str

Identifier of the folder.

required

Returns:

Type Description
FolderUpdate

The resulting FolderUpdate.

Source code in src/pyromax/core/CoreMixins/User.py
171
172
173
174
175
176
177
178
179
async def delete_folder(self, folder_id: str) -> FolderUpdate:
    """Delete folder.

    :param folder_id: Identifier of the folder.
    :type folder_id: str
    :returns: The resulting FolderUpdate.
    :rtype: FolderUpdate
    """
    return await self.delete_folders(folder_ids=[folder_id])

close_all_sessions async

close_all_sessions() -> bool

Close all sessions.

Returns:

Type Description
bool

True if the server accepted the request; otherwise False.

Source code in src/pyromax/core/CoreMixins/User.py
181
182
183
184
185
186
187
188
189
190
191
192
async def close_all_sessions(self) -> bool:
    """Close all sessions.

    :returns: ``True`` if the server accepted the request; otherwise ``False``.
    :rtype: bool
    """
    return cast(
        bool,
        await self(
            CloseAllSessionsMethod,
        ),
    )

logout async

logout() -> None

Logout.

Source code in src/pyromax/core/CoreMixins/User.py
194
195
196
async def logout(self) -> None:
    """Logout."""
    return cast(None, await self(LogoutMethod))

set_presence async

set_presence(online: bool) -> None

Set presence.

Parameters:

Name Type Description Default
online bool

The online value.

required
Source code in src/pyromax/core/CoreMixins/User.py
198
199
200
201
202
203
204
205
206
207
208
209
210
async def set_presence(self, online: bool) -> None:
    """Set presence.

    :param online: The online value.
    :type online: bool
    """
    return cast(
        None,
        await self(
            SetPresenceMethod,
            online=online,
        ),
    )

change_profile_settings async

change_profile_settings(
    privacy_settings: PrivacySettings,
) -> None

Update the account privacy settings.

Parameters:

Name Type Description Default
privacy_settings PrivacySettings

Privacy settings to apply to the account.

required

Raises:

Type Description
ValueError

If updating the privacy settings fails.

Source code in src/pyromax/core/CoreMixins/User.py
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
async def change_profile_settings(self, privacy_settings: PrivacySettings) -> None:
    """Update the account privacy settings.

    :param privacy_settings: Privacy settings to apply to the account.

    :raises ValueError: If updating the privacy settings fails.

    :type privacy_settings: PrivacySettings
    """

    try:
        return cast(
            None,
            await self(
                ChangeProfileSettingsMethod,
                privacy_settings=privacy_settings,
            ),
        )
    except MapperApiError as e:
        raise ValueError("Change profile settings failed") from e

create_group async

create_group(
    title: str,
    participant_ids: list[int] | None = None,
    notify: bool = True,
    chat_type: str = "CHAT",
    event: str = "new",
    typeof: str = "CONTROL",
) -> tuple[Chat, Message] | tuple[None, None]

Create group.

Parameters:

Name Type Description Default
title str

The title value.

required
participant_ids list[int] | None

Identifiers of the participant objects.

None
notify bool

Whether MAX should notify affected users.

True
chat_type str

The chat type value.

'CHAT'
event str

Incoming event to process.

'new'
typeof str

Attachment class that determines the upload type.

'CONTROL'

Returns:

Type Description
tuple[Chat, Message] | tuple[None, None]

The resulting tuple[Chat, Message] | tuple[None, None] value.

Source code in src/pyromax/core/CoreMixins/Chat.py
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
async def create_group(
    self,
    title: str,
    participant_ids: list[int] | None = None,
    notify: bool = True,
    chat_type: str = "CHAT",
    event: str = "new",
    typeof: str = "CONTROL",
) -> tuple[Chat, Message] | tuple[None, None]:
    """Create group.

    :param title: The title value.
    :type title: str
    :param participant_ids: Identifiers of the participant objects.
    :type participant_ids: list[int] | None
    :param notify: Whether MAX should notify affected users.
    :type notify: bool
    :param chat_type: The chat type value.
    :type chat_type: str
    :param event: Incoming event to process.
    :type event: str
    :param typeof: Attachment class that determines the upload type.
    :type typeof: str
    :returns: The resulting tuple[Chat, Message] | tuple[None, None] value.
    :rtype: tuple[Chat, Message] | tuple[None, None]
    """

    return cast(
        tuple[Chat, Message] | tuple[None, None],
        await self(
            CreateGroupMethod,
            title=title,
            participant_ids=participant_ids,
            notify=notify,
            chat_type=chat_type,
            event=event,
            typeof=typeof,
        ),
    )

invite_users_to_group async

invite_users_to_group(
    chat_id: int,
    user_ids: list[int],
    show_history: bool = True,
) -> Chat | None

Invite users to group.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
user_ids list[int]

Identifiers of the users.

required
show_history bool

Show message history to new members.

True

Returns:

Type Description
Chat | None

The resulting Chat | None value.

Source code in src/pyromax/core/CoreMixins/Chat.py
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
async def invite_users_to_group(
    self,
    chat_id: int,
    user_ids: list[int],
    show_history: bool = True,
) -> Chat | None:
    """Invite users to group.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param user_ids: Identifiers of the users.
    :type user_ids: list[int]
    :param show_history: Show message history to new members.
    :type show_history: bool
    :returns: The resulting Chat | None value.
    :rtype: Chat | None
    """

    return cast(
        Chat | None,
        await self(
            InviteUsersToGroupMethod,
            chat_id=chat_id,
            user_ids=user_ids,
            show_history=show_history,
        ),
    )

remove_users_from_group async

remove_users_from_group(
    chat_id: int,
    user_ids: list[int] | list[str],
    clean_msg_period: int,
) -> Chat | None

Remove users from group.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
user_ids list[int] | list[str]

Identifiers of the users.

required
clean_msg_period int

Cleanup period for messages from removed participants.

required

Returns:

Type Description
Chat | None

The resulting Chat | None value.

Source code in src/pyromax/core/CoreMixins/Chat.py
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
async def remove_users_from_group(
    self,
    chat_id: int,
    user_ids: list[int] | list[str],
    clean_msg_period: int,
) -> Chat | None:
    """Remove users from group.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param user_ids: Identifiers of the users.
    :type user_ids: list[int] | list[str]
    :param clean_msg_period: Cleanup period for messages from removed participants.
    :type clean_msg_period: int
    :returns: The resulting Chat | None value.
    :rtype: Chat | None
    """

    return cast(
        Chat | None,
        await self(
            RemoveUsersFromGroupMethod,
            chat_id=chat_id,
            user_ids=user_ids,
            clean_msg_period=clean_msg_period,
        ),
    )

change_group_settings async

change_group_settings(
    chat_id: int,
    all_can_pin_message: bool | None = None,
    only_owner_can_change_icon_title: bool | None = None,
    only_admin_can_add_member: bool | None = None,
    only_admin_can_call: bool | None = None,
    members_can_see_private_link: bool | None = None,
) -> Chat | None

Change group settings.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
all_can_pin_message bool | None

All participants can pin messages.

None
only_owner_can_change_icon_title bool | None

The only owner can change icon title.

None
only_admin_can_add_member bool | None

The only admin can add member.

None
only_admin_can_call bool | None

The only admin can call.

None
members_can_see_private_link bool | None

The members can see private link.

None

Returns:

Type Description
Chat | None

The resulting Chat | None value.

Source code in src/pyromax/core/CoreMixins/Chat.py
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
async def change_group_settings(
    self,
    chat_id: int,
    all_can_pin_message: bool | None = None,
    only_owner_can_change_icon_title: bool | None = None,
    only_admin_can_add_member: bool | None = None,
    only_admin_can_call: bool | None = None,
    members_can_see_private_link: bool | None = None,
) -> Chat | None:
    """Change group settings.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param all_can_pin_message: All participants can pin messages.
    :type all_can_pin_message: bool | None
    :param only_owner_can_change_icon_title: The only owner can change icon title.
    :type only_owner_can_change_icon_title: bool | None
    :param only_admin_can_add_member: The only admin can add member.
    :type only_admin_can_add_member: bool | None
    :param only_admin_can_call: The only admin can call.
    :type only_admin_can_call: bool | None
    :param members_can_see_private_link: The members can see private link.
    :type members_can_see_private_link: bool | None
    :returns: The resulting Chat | None value.
    :rtype: Chat | None
    """

    return cast(
        Chat | None,
        await self(
            ChangeGroupSettingsMethod,
            chat_id=chat_id,
            all_can_pin_message=all_can_pin_message,
            only_owner_can_change_icon_title=only_owner_can_change_icon_title,
            only_admin_can_add_member=only_admin_can_add_member,
            only_admin_can_call=only_admin_can_call,
            members_can_see_private_link=members_can_see_private_link,
        ),
    )

change_group_profile async

change_group_profile(
    chat_id: int,
    name: str | None = None,
    description: str | None = None,
) -> Chat | None

Change group profile.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
name str | None

The name of the chat.

None
description str | None

The description of the chat.

None

Returns:

Type Description
Chat | None

The resulting Chat | None value.

Source code in src/pyromax/core/CoreMixins/Chat.py
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
async def change_group_profile(
    self,
    chat_id: int,
    name: str | None = None,
    description: str | None = None,
) -> Chat | None:
    """Change group profile.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param name: The name of the chat.
    :type name: str | None
    :param description: The description of the chat.
    :type description: str | None
    :returns: The resulting Chat | None value.
    :rtype: Chat | None
    """

    return cast(
        Chat | None,
        await self(
            ChangeGroupProfileMethod,
            chat_id=chat_id,
            name=name,
            description=description,
        ),
    )

join_group async

join_group(link: str) -> Chat

Join group.

Parameters:

Name Type Description Default
link str

Invite group link.

required

Returns:

Type Description
Chat

The resulting Chat value.

Source code in src/pyromax/core/CoreMixins/Chat.py
198
199
200
201
202
203
204
205
206
207
208
209
210
async def join_group(self, link: str) -> Chat:
    """Join group.

    :param link: Invite group link.
    :type link: str
    :returns: The resulting Chat value.
    :rtype: Chat
    """

    return cast(
        Chat,
        await self(JoinGroupMethod, link=link),
    )

join_channel async

join_channel(link: str) -> Chat

Join channel.

Parameters:

Name Type Description Default
link str

Invite channel link.

required

Returns:

Type Description
Chat

The resulting Chat value.

Source code in src/pyromax/core/CoreMixins/Chat.py
212
213
214
215
216
217
218
219
220
221
222
223
224
async def join_channel(self, link: str) -> Chat:
    """Join channel.

    :param link: Invite channel link.
    :type link: str
    :returns: The resulting Chat value.
    :rtype: Chat
    """

    return cast(
        Chat,
        await self(JoinChannelMethod, link=link),
    )
resolve_group_by_link(link: str) -> Chat | None

Resolve group by link.

Parameters:

Name Type Description Default
link str

Invite group link.

required

Returns:

Type Description
Chat | None

The resulting Chat | None value.

Source code in src/pyromax/core/CoreMixins/Chat.py
226
227
228
229
230
231
232
233
234
235
236
237
238
async def resolve_group_by_link(self, link: str) -> Chat | None:
    """Resolve group by link.

    :param link: Invite group link.
    :type link: str
    :returns: The resulting Chat | None value.
    :rtype: Chat | None
    """

    return cast(
        Chat | None,
        await self(ResolveGroupByLinkMethod, link=link),
    )
revoke_invite_link(chat_id: int) -> Chat

Revoke invite link.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required

Returns:

Type Description
Chat

The resulting Chat value.

Source code in src/pyromax/core/CoreMixins/Chat.py
240
241
242
243
244
245
246
247
248
249
250
251
252
async def revoke_invite_link(self, chat_id: int) -> Chat:
    """Revoke invite link.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :returns: The resulting Chat value.
    :rtype: Chat
    """

    return cast(
        Chat,
        await self(RevokeInviteLinkMethod, chat_id=chat_id),
    )

get_chats async

get_chats(chat_ids: Iterable[int]) -> list[Chat]

Retrieve chats.

Parameters:

Name Type Description Default
chat_ids Iterable[int]

Identifiers of the chats.

required

Returns:

Type Description
list[Chat]

The collection from gotten chats.

Source code in src/pyromax/core/CoreMixins/Chat.py
254
255
256
257
258
259
260
261
262
263
264
265
266
async def get_chats(self, chat_ids: Iterable[int]) -> list[Chat]:
    """Retrieve chats.

    :param chat_ids: Identifiers of the chats.
    :type chat_ids: Iterable[int]
    :returns: The collection from gotten chats.
    :rtype: list[Chat]
    """

    return cast(
        list[Chat],
        await self(GetChatsMethod, chat_ids=chat_ids),
    )

get_chat async

get_chat(chat_id: int) -> Chat

Retrieve chat.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required

Returns:

Type Description
Chat

The resulting Chat value.

Raises:

Type Description
ValueError

If chat not found.

Source code in src/pyromax/core/CoreMixins/Chat.py
268
269
270
271
272
273
274
275
276
277
278
279
280
async def get_chat(self, chat_id: int) -> Chat:
    """Retrieve chat.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :returns: The resulting Chat value.
    :rtype: Chat
    :raises ValueError: If chat not found.
    """
    chats = await self.get_chats([chat_id])
    if not chats:
        raise ValueError("Chat not found")
    return chats[0]

leave_group async

leave_group(chat_id: int) -> Message | None

Leave group.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required

Returns:

Type Description
Message | None

The resulting Message | None value.

Source code in src/pyromax/core/CoreMixins/Chat.py
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
async def leave_group(self, chat_id: int) -> Message | None:
    """Leave group.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :returns: The resulting Message | None value.
    :rtype: Message | None
    """

    return cast(
        Message | None,
        await self(
            LeaveGroupMethod,
            chat_id=chat_id,
        ),
    )

leave_channel async

leave_channel(chat_id: int) -> Message | None

Leave channel.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required

Returns:

Type Description
Message | None

The resulting Message | None value.

Source code in src/pyromax/core/CoreMixins/Chat.py
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
async def leave_channel(self, chat_id: int) -> Message | None:
    """Leave channel.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :returns: The resulting Message | None value.
    :rtype: Message | None
    """

    return cast(
        Message | None,
        await self(
            LeaveChannelMethod,
            chat_id=chat_id,
        ),
    )

fetch_chats async

fetch_chats(marker: int | None = None) -> list[Chat]

Fetch chats.

Parameters:

Name Type Description Default
marker int | None

Pagination marker in milliseconds. If None, the current time is used.

None

Returns:

Type Description
list[Chat]

The resulting Chats collection.

Source code in src/pyromax/core/CoreMixins/Chat.py
316
317
318
319
320
321
322
323
324
325
326
327
328
async def fetch_chats(self, marker: int | None = None) -> list[Chat]:
    """Fetch chats.

    :param marker: Pagination marker in milliseconds. If ``None``, the current time is used.
    :type marker: int | None
    :returns: The resulting Chats collection.
    :rtype: list[Chat]
    """

    return cast(
        list[Chat],
        await self(FetchChatsMethod, marker=marker),
    )

get_join_requests async

get_join_requests(
    chat_id: int, count: int = 100
) -> list[Member]

Retrieve join requests.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
count int

Maximum number of items to retrieve.

100

Returns:

Type Description
list[Member]

The resulting Members collection.

Source code in src/pyromax/core/CoreMixins/Chat.py
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
async def get_join_requests(self, chat_id: int, count: int = 100) -> list[Member]:
    """Retrieve join requests.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param count: Maximum number of items to retrieve.
    :type count: int
    :returns: The resulting Members collection.
    :rtype: list[Member]
    """

    return cast(
        list[Member],
        await self(
            GetJoinRequestsMethod,
            chat_id=chat_id,
            count=count,
        ),
    )

confirm_join_requests async

confirm_join_requests(
    chat_id: int,
    user_ids: Iterable[int],
    show_history: bool = True,
) -> Chat | None

Confirm join requests.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
user_ids Iterable[int]

Identifiers of the users.

required
show_history bool

Show message history to new members.

True

Returns:

Type Description
Chat | None

The resulting Chat | None value.

Source code in src/pyromax/core/CoreMixins/Chat.py
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
async def confirm_join_requests(
    self,
    chat_id: int,
    user_ids: Iterable[int],
    show_history: bool = True,
) -> Chat | None:
    """Confirm join requests.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param user_ids: Identifiers of the users.
    :type user_ids: Iterable[int]
    :param show_history: Show message history to new members.
    :type show_history: bool
    :returns: The resulting Chat | None value.
    :rtype: Chat | None
    """

    return cast(
        Chat | None,
        await self(
            ConfirmJoinRequestsMethod,
            chat_id=chat_id,
            user_ids=user_ids,
            show_history=show_history,
        ),
    )

confirm_join_request async

confirm_join_request(
    chat_id: int, user_id: int, show_history: bool = True
) -> Chat | None

Confirm join request.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
user_id int

Identifier of the user.

required
show_history bool

Show message history to new members.

True

Returns:

Type Description
Chat | None

The resulting Chat | None value.

Source code in src/pyromax/core/CoreMixins/Chat.py
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
async def confirm_join_request(
    self,
    chat_id: int,
    user_id: int,
    show_history: bool = True,
) -> Chat | None:
    """Confirm join request.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param user_id: Identifier of the user.
    :type user_id: int
    :param show_history: Show message history to new members.
    :type show_history: bool
    :returns: The resulting Chat | None value.
    :rtype: Chat | None
    """
    return await self.confirm_join_requests(
        chat_id=chat_id,
        user_ids=[user_id],
        show_history=show_history,
    )

decline_join_requests async

decline_join_requests(
    chat_id: int, user_ids: Iterable[int]
) -> Chat | None

Decline join requests.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
user_ids Iterable[int]

Identifiers of the users.

required

Returns:

Type Description
Chat | None

The resulting Chat | None value.

Source code in src/pyromax/core/CoreMixins/Chat.py
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
async def decline_join_requests(
    self,
    chat_id: int,
    user_ids: Iterable[int],
) -> Chat | None:
    """Decline join requests.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param user_ids: Identifiers of the users.
    :type user_ids: Iterable[int]
    :returns: The resulting Chat | None value.
    :rtype: Chat | None
    """

    return cast(
        Chat | None,
        await self(
            DeclineJoinRequestsMethod,
            chat_id=chat_id,
            user_ids=user_ids,
        ),
    )

decline_join_request async

decline_join_request(
    chat_id: int, user_id: int
) -> Chat | None

Decline join request.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
user_id int

Identifier of the user.

required

Returns:

Type Description
Chat | None

The resulting Chat | None value.

Source code in src/pyromax/core/CoreMixins/Chat.py
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
async def decline_join_request(
    self,
    chat_id: int,
    user_id: int,
) -> Chat | None:
    """Decline join request.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param user_id: Identifier of the user.
    :type user_id: int
    :returns: The resulting Chat | None value.
    :rtype: Chat | None
    """
    return await self.decline_join_requests(
        chat_id=chat_id,
        user_ids=[user_id],
    )

delete_chat async

delete_chat(
    chat_id: int,
    last_event_time: int | None = None,
    for_all: bool = True,
) -> None

Delete chat.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
last_event_time int | None

The last event time value.

None
for_all bool

Delete only for the current account.

True
Source code in src/pyromax/core/CoreMixins/Chat.py
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
async def delete_chat(
    self,
    chat_id: int,
    last_event_time: int | None = None,
    for_all: bool = True,
) -> None:
    """Delete chat.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param last_event_time: The last event time value.
    :type last_event_time: int | None
    :param for_all: Delete only for the current account.
    :type for_all: bool
    """
    return cast(
        None,
        await self(
            DeleteChatMethod,
            chat_id=chat_id,
            last_event_time=last_event_time,
            for_all=for_all,
        ),
    )

add_admin async

add_admin(
    chat_id: int,
    user_id: int,
    permissions: Iterable[ChannelPermissions],
) -> None

Add admin.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
user_id int

Identifier of the user.

required
permissions Iterable[ChannelPermissions]

Collection of permissions.

required
Source code in src/pyromax/core/CoreMixins/Chat.py
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
async def add_admin(
    self, chat_id: int, user_id: int, permissions: Iterable[ChannelPermissions]
) -> None:
    """Add admin.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param user_id: Identifier of the user.
    :type user_id: int
    :param permissions: Collection of permissions.
    :type permissions: Iterable[ChannelPermissions]
    """
    return cast(
        None,
        await self(
            AddAdminMethod,
            chat_id=chat_id,
            user_id=user_id,
            permissions=permissions,
        ),
    )

send_message async

send_message(
    chat_id: int,
    text: str = "",
    attaches: list[BaseFileAttachment] | None = None,
    link: MessageLink | None = None,
    notify: bool = True,
) -> Message | None

Send a message to a chat.

Parameters:

Name Type Description Default
chat_id int

Target chat identifier.

required
text str

Message text.

''
attaches list[BaseFileAttachment] | None

Optional list of attachments.

None
link MessageLink | None

Optional message link object.

None
notify bool

Whether MAX should notify affected users.

True

Returns:

Type Description
Message | None

API response returned by the mapper.

Raises:

Type Description
SendMessageError

If message sending fails.

AttributeError

If logger not initialized in MaxApi instance.

Source code in src/pyromax/core/CoreMixins/Message.py
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
79
80
async def send_message(
    self,
    chat_id: int,
    text: str = "",
    attaches: list[BaseFileAttachment] | None = None,
    link: MessageLink | None = None,
    notify: bool = True,
) -> Message | None:
    """Send a message to a chat.

    :param chat_id: Target chat identifier.
    :type chat_id: int
    :param text: Message text.
    :type text: str
    :param attaches: Optional list of attachments.
    :type attaches: list[BaseFileAttachment] | None
    :param link: Optional message link object.
    :type link: MessageLink | None

    :returns: API response returned by the mapper.
    :rtype: Message | None

    :raises SendMessageError: If message sending fails.

    :param notify: Whether MAX should notify affected users.
    :type notify: bool
    :raises AttributeError: If logger not initialized in MaxApi instance.
    """

    try:
        return cast(
            Message | None,
            await self(
                SendMessageMethod,
                text=text,
                chat_id=chat_id,
                attaches=attaches,
                link=link,
                notify=notify,
            ),
        )
    except SendMessageError as e:
        if self._logger is None:
            raise AttributeError("logger not initialized in MaxApi instance")
        self._logger.warning("Failed to send message: %s", e)
        raise e

forward_message async

forward_message(
    message_id: int | str,
    to_chat_id: int,
    from_chat_id: int,
    notify: bool = True,
) -> Message | None

Forward message.

Parameters:

Name Type Description Default
message_id int | str

Identifier of the message.

required
to_chat_id int

Identifier of the destination chat.

required
from_chat_id int

Identifier of the source chat.

required
notify bool

Whether MAX should notify affected users.

True

Returns:

Type Description
Message | None

The resulting Message | None value.

Raises:

Type Description
AttributeError

If logger not initialized in MaxApi instance.

Source code in src/pyromax/core/CoreMixins/Message.py
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
async def forward_message(
    self,
    message_id: int | str,
    to_chat_id: int,
    from_chat_id: int,
    notify: bool = True,
) -> Message | None:
    """Forward message.

    :param message_id: Identifier of the message.
    :type message_id: int | str
    :param to_chat_id: Identifier of the destination chat.
    :type to_chat_id: int
    :param from_chat_id: Identifier of the source chat.
    :type from_chat_id: int
    :param notify: Whether MAX should notify affected users.
    :type notify: bool
    :returns: The resulting Message | None value.
    :rtype: Message | None
    :raises AttributeError: If logger not initialized in MaxApi instance.
    """
    try:
        return cast(
            Message | None,
            await self(
                ForwardMessageMethod,
                message_id=message_id,
                to_chat_id=to_chat_id,
                from_chat_id=from_chat_id,
                notify=notify,
            ),
        )
    except SendMessageError as e:
        if self._logger is None:
            raise AttributeError("logger not initialized in MaxApi instance")
        self._logger.warning("Failed to forward message: %s", e)
        raise e

edit_message async

edit_message(
    chat_id: int,
    message_id: int | str,
    text: str | None = None,
    attaches: list[BaseFileAttachment] | None = None,
    **kwargs: Any
) -> Message

Edit message.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
message_id int | str

Identifier of the message.

required
text str | None

Message or textual content.

None
attaches list[BaseFileAttachment] | None

Attachments associated with the message.

None

Returns:

Type Description
Message

The resulting Message value.

Source code in src/pyromax/core/CoreMixins/Message.py
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
async def edit_message(
    self,
    chat_id: int,
    message_id: int | str,
    text: str | None = None,
    attaches: list[BaseFileAttachment] | None = None,
    **kwargs: Any,
) -> Message:
    """Edit message.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param message_id: Identifier of the message.
    :type message_id: int | str
    :param text: Message or textual content.
    :type text: str | None
    :param attaches: Attachments associated with the message.
    :type attaches: list[BaseFileAttachment] | None
    # :param kwargs: Keyword arguments forwarded to the wrapped callable.
    # :type kwargs: Any
    :returns: The resulting Message value.
    :rtype: Message
    """

    return cast(
        Message,
        await self(
            EditMessageMethod,
            chat_id=chat_id,
            message_id=message_id,
            text=text,
            attaches=attaches,
        ),
    )

get_messages async

get_messages(
    chat_id: int, message_ids: Iterable[str] | Iterable[int]
) -> list[Message]

Retrieve messages.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
message_ids Iterable[str] | Iterable[int]

Identifiers of the messages.

required

Returns:

Type Description
list[Message]

The resulting collection.

Source code in src/pyromax/core/CoreMixins/Message.py
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
async def get_messages(
    self, chat_id: int, message_ids: Iterable[str] | Iterable[int]
) -> list[Message]:
    """Retrieve messages.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param message_ids: Identifiers of the messages.
    :type message_ids: Iterable[str] | Iterable[int]
    :returns: The resulting collection.
    :rtype: list[Message]
    """

    return cast(
        list[Message],
        await self(
            GetMessagesMethod,
            chat_id=chat_id,
            message_ids=message_ids,
        ),
    )

get_message async

get_message(
    chat_id: int, message_id: int | str
) -> Message | None

Retrieve message.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
message_id int | str

Identifier of the message.

required

Returns:

Type Description
Message | None

The resulting Message | None value.

Source code in src/pyromax/core/CoreMixins/Message.py
177
178
179
180
181
182
183
184
185
186
187
188
async def get_message(self, chat_id: int, message_id: int | str) -> Message | None:
    """Retrieve message.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param message_id: Identifier of the message.
    :type message_id: int | str
    :returns: The resulting Message | None value.
    :rtype: Message | None
    """
    msgs = await self.get_messages(chat_id=chat_id, message_ids=[message_id])
    return msgs[0] if msgs else None

get_chat_history async

get_chat_history(
    chat_id: int,
    forward: int = ...,
    backward: int = ...,
    backward_time: int = ...,
    forward_time: int = ...,
    from_time: int | None = ...,
    item_type: Literal["DELAYED", "REGULAR"] = ...,
    get_chat: bool = ...,
    get_messages: Literal[True] = True,
    interactive: bool = ...,
) -> list[Message]
get_chat_history(
    chat_id: int,
    forward: int = ...,
    backward: int = ...,
    backward_time: int = ...,
    forward_time: int = ...,
    from_time: int | None = None,
    item_type: Literal["DELAYED", "REGULAR"] = ...,
    get_chat: bool = ...,
    get_messages: Literal[False] = False,
    interactive: bool = ...,
) -> list[str]
get_chat_history(
    chat_id: int,
    forward: int = ...,
    backward: int = ...,
    backward_time: int = ...,
    forward_time: int = ...,
    from_time: int | None = ...,
    item_type: Literal["DELAYED", "REGULAR"] = ...,
    get_chat: bool = ...,
    get_messages: bool = ...,
    interactive: bool = ...,
) -> list[Message] | list[str]
get_chat_history(
    chat_id: int,
    forward: int = 0,
    backward: int = 40,
    backward_time: int = 0,
    forward_time: int = 0,
    from_time: int | None = None,
    item_type: Literal["DELAYED", "REGULAR"] = "REGULAR",
    get_chat: bool = False,
    get_messages: bool = True,
    interactive: bool = False,
) -> list[Message] | list[str]

Retrieve chat history.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
forward int

How many messages to load ahead from from_time.

0
backward int

How many messages to load back from from_time.

40
backward_time int

Look-back time window in milliseconds.

0
forward_time int

Forward time window in milliseconds.

0
from_time int | None

The reference point in Unix time (milliseconds). If None, the current moment is used.

None
item_type Literal['DELAYED', 'REGULAR']

History item type.

'REGULAR'
get_chat bool

Request chat data along with the history.

False
get_messages bool

The get messages value.

True
interactive bool

Request the messages themselves.

False

Returns:

Type Description
list[Message] | list[str]

Message collection if get_messages is True else message ids collection.

Source code in src/pyromax/core/CoreMixins/Message.py
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
async def get_chat_history(
    self,
    chat_id: int,
    forward: int = 0,
    backward: int = 40,
    backward_time: int = 0,
    forward_time: int = 0,
    from_time: int | None = None,
    item_type: Literal["DELAYED", "REGULAR"] = "REGULAR",
    get_chat: bool = False,
    get_messages: bool = True,
    interactive: bool = False,
) -> list[Message] | list[str]:
    """Retrieve chat history.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param forward: How many messages to load ahead from ``from_time``.
    :type forward: int
    :param backward: How many messages to load back from ``from_time``.
    :type backward: int
    :param backward_time: Look-back time window in milliseconds.
    :type backward_time: int
    :param forward_time: Forward time window in milliseconds.
    :type forward_time: int
    :param from_time: The reference point in Unix time (milliseconds). If ``None``, the current moment is used.
    :type from_time: int | None
    :param item_type: History item type.
    :type item_type: Literal['DELAYED', 'REGULAR']
    :param get_chat: Request chat data along with the history.
    :type get_chat: bool
    :param get_messages: The get messages value.
    :type get_messages: bool
    :param interactive: Request the messages themselves.
    :type interactive: bool
    :returns: Message collection if get_messages is True else message ids collection.
    :rtype: list[Message] | list[str]
    """

    return cast(
        list[Message] | list[str],
        await self(
            GetChatHistoryMethod,
            chat_id=chat_id,
            forward=forward,
            backward=backward,
            backward_time=backward_time,
            forward_time=forward_time,
            from_time=from_time,
            item_type=item_type,
            get_chat=get_chat,
            get_messages=get_messages,
            interactive=interactive,
        ),
    )

delete_messages async

delete_messages(
    chat_id: int,
    message_ids: list[str] | list[int],
    for_me: bool = False,
) -> None

Delete messages.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
message_ids list[str] | list[int]

Identifiers of the messages.

required
for_me bool

Delete only for the current account.

False
Source code in src/pyromax/core/CoreMixins/Message.py
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
async def delete_messages(
    self,
    chat_id: int,
    message_ids: list[str] | list[int],
    for_me: bool = False,
) -> None:
    """Delete messages.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param message_ids: Identifiers of the messages.
    :type message_ids: list[str] | list[int]
    :param for_me: Delete only for the current account.
    :type for_me: bool
    """
    return cast(
        None,
        await self(
            DeleteMessagesMethod,
            chat_id=chat_id,
            message_ids=message_ids,
            for_me=for_me,
        ),
    )

pin_message async

pin_message(
    chat_id: int, message_id: int | str, notify: bool = True
) -> None

Pin message.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
message_id int | str

Identifier of the message.

required
notify bool

Whether MAX should notify affected users.

True
Source code in src/pyromax/core/CoreMixins/Message.py
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
async def pin_message(
    self, chat_id: int, message_id: int | str, notify: bool = True
) -> None:
    """Pin message.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param message_id: Identifier of the message.
    :type message_id: int | str
    :param notify: Whether MAX should notify affected users.
    :type notify: bool
    """
    return cast(
        None,
        await self(
            PinMessageMethod,
            chat_id=chat_id,
            message_id=message_id,
            notify=notify,
        ),
    )

add_reaction async

add_reaction(
    chat_id: int,
    message_id: int | str,
    reaction_id: str,
    reaction_type: str = "EMOJI",
) -> EmojiReaction | None

Add an emoji reaction to a message.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat containing the message.

required
message_id int | str

int | str

required
reaction_id str

str

required
reaction_type str

str

'EMOJI'

Returns:

Type Description
EmojiReaction | None

info about reaction or None if cannot get this info

Raises:

Type Description
ReactionError

if adding reaction failed

Source code in src/pyromax/core/CoreMixins/Message.py
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
async def add_reaction(
    self,
    chat_id: int,
    message_id: int | str,
    reaction_id: str,
    reaction_type: str = "EMOJI",
) -> EmojiReaction | None:
    """Add an emoji reaction to a message.

    :param chat_id: Identifier of the chat containing the message.
    :type chat_id: int
    :param message_id: int | str
    :type message_id: int | str
    :param reaction_id: str
    :type reaction_id: str
    :param reaction_type: str
    :type reaction_type: str

    :returns: info about reaction or None if cannot get this info
    :rtype: EmojiReaction | None

    :raises ReactionError: if adding reaction failed
    """

    try:
        return cast(
            EmojiReaction | None,
            await self(
                AddReactionMethod,
                chat_id=chat_id,
                message_id=message_id,
                reaction_id=reaction_id,
                reaction_type=reaction_type,
            ),
        )
    except MapperApiError as e:
        raise ReactionError("add reaction failed") from e

remove_reaction async

remove_reaction(
    chat_id: int, message_id: int | str
) -> EmojiReaction | None

Remove reaction.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
message_id int | str

Identifier of the message.

required

Returns:

Type Description
EmojiReaction | None

The resulting EmojiReaction | None value.

Source code in src/pyromax/core/CoreMixins/Message.py
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
async def remove_reaction(
    self,
    chat_id: int,
    message_id: int | str,
) -> EmojiReaction | None:
    """Remove reaction.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param message_id: Identifier of the message.
    :type message_id: int | str
    :returns: The resulting EmojiReaction | None value.
    :rtype: EmojiReaction | None
    """

    return cast(
        EmojiReaction | None,
        await self(RemoveReactionMethod, chat_id=chat_id, message_id=message_id),
    )

get_reactions async

get_reactions(
    chat_id: int, message_ids: list[str] | list[int]
) -> dict[str, EmojiReaction] | None

Retrieve reactions.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
message_ids list[str] | list[int]

Identifiers of the messages.

required

Returns:

Type Description
dict[str, EmojiReaction] | None

The resulting dict[, ] | None value.

Source code in src/pyromax/core/CoreMixins/Message.py
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
async def get_reactions(
    self,
    chat_id: int,
    message_ids: list[str] | list[int],
) -> dict[str, EmojiReaction] | None:
    """Retrieve reactions.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param message_ids: Identifiers of the messages.
    :type message_ids: list[str] | list[int]
    :returns: The resulting dict[<message_id>, <EmojiReaction for this message>] | None value.
    :rtype: dict[str, EmojiReaction] | None
    """

    return cast(
        dict[str, EmojiReaction] | None,
        await self(GetReactionsMethod, chat_id=chat_id, message_ids=message_ids),
    )

read_message async

read_message(
    chat_id: int,
    message_id: int | str,
    typeof: str = "READ_MESSAGE",
    mark: int | None = None,
) -> ReadState

Read message.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
message_id int | str

Identifier of the message.

required
typeof str

Attachment class that determines the upload type.

'READ_MESSAGE'
mark int | None

The mark value.

None

Returns:

Type Description
ReadState

The resulting ReadState value.

Source code in src/pyromax/core/CoreMixins/Message.py
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
async def read_message(
    self,
    chat_id: int,
    message_id: int | str,
    typeof: str = "READ_MESSAGE",
    mark: int | None = None,
) -> ReadState:
    """Read message.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param message_id: Identifier of the message.
    :type message_id: int | str
    :param typeof: Attachment class that determines the upload type.
    :type typeof: str
    :param mark: The mark value.
    :type mark: int | None
    :returns: The resulting ReadState value.
    :rtype: ReadState
    """

    return cast(
        ReadState,
        await self(
            ReadMessageMethod,
            chat_id=chat_id,
            message_id=message_id,
            typeof=typeof,
            mark=mark,
        ),
    )

create_poll async

create_poll(poll: Poll) -> Poll

Create poll.

Parameters:

Name Type Description Default
poll Poll

Poll instance to process.

required

Returns:

Type Description
Poll

The resulting Poll value what can be send with message in send_message method.

Source code in src/pyromax/core/CoreMixins/Message.py
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
async def create_poll(
    self,
    poll: Poll,
) -> Poll:
    """Create poll.

    :param poll: Poll instance to process.
    :type poll: Poll
    :returns: The resulting Poll value what can be send with message in send_message method.
    :rtype: Poll
    """

    return cast(
        Poll,
        await self(
            CreatePollMethod,
            poll=poll,
        ),
    )

vote_poll async

vote_poll(
    chat_id: int,
    message_id: int | str,
    poll_id: int,
    answer_ids: list[int],
) -> PollState

Submit a vote for poll.

Parameters:

Name Type Description Default
chat_id int

Identifier of the chat.

required
message_id int | str

Identifier of the message.

required
poll_id int

Identifier of the poll.

required
answer_ids list[int]

Identifiers of the answer objects.

required

Returns:

Type Description
PollState

The resulting PollState value.

Source code in src/pyromax/core/CoreMixins/Message.py
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
async def vote_poll(
    self,
    chat_id: int,
    message_id: int | str,
    poll_id: int,
    answer_ids: list[int],
) -> PollState:
    """Submit a vote for poll.

    :param chat_id: Identifier of the chat.
    :type chat_id: int
    :param message_id: Identifier of the message.
    :type message_id: int | str
    :param poll_id: Identifier of the poll.
    :type poll_id: int
    :param answer_ids: Identifiers of the answer objects.
    :type answer_ids: list[int]
    :returns: The resulting PollState value.
    :rtype: PollState
    """

    return cast(
        PollState,
        await self(
            VotePollMethod,
            chat_id=chat_id,
            message_id=message_id,
            poll_id=poll_id,
            answer_ids=answer_ids,
        ),
    )

set_2fa async

set_2fa(
    password: str,
    email: str | None = None,
    hint: str | None = None,
    email_code_getter: (
        Callable[[str], Coroutine[Any, Any, str]] | None
    ) = None,
    two_factor_actions: list[TwoFactorAction] | None = None,
) -> None

Set 2fa.

Parameters:

Name Type Description Default
password str

New 2FA password.

required
email str | None

Email address for 2FA, if required.

None
hint str | None

Password hint, if required.

None
email_code_getter Callable[[str], Coroutine[Any, Any, str]] | None

Callable to get password, first argument is phone number.

None
two_factor_actions list[TwoFactorAction] | None

Collection of two factor actions.

None
Source code in src/pyromax/core/CoreMixins/Auth.py
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
async def set_2fa(
    self,
    password: str,
    email: str | None = None,
    hint: str | None = None,
    email_code_getter: Callable[[str], Coroutine[Any, Any, str]] | None = None,
    two_factor_actions: list[TwoFactorAction] | None = None,
) -> None:
    """Set 2fa.

    :param password: New 2FA password.
    :type password: str
    :param email: Email address for 2FA, if required.
    :type email: str | None
    :param hint: Password hint, if required.
    :type hint: str | None
    :param email_code_getter: Callable to get password, first argument is phone number.
    :type email_code_getter: Callable[[str], Coroutine[Any, Any, str]] | None
    :param two_factor_actions: Collection of two factor actions.
    :type two_factor_actions: list[TwoFactorAction] | None
    """
    return cast(
        None,
        await self(
            Set2FaMethod,
            password=password,
            email=email,
            hint=hint,
            email_code_getter=email_code_getter,
            two_factor_actions=two_factor_actions,
        ),
    )

remove_2fa async

remove_2fa(
    password: str,
    two_factor_actions: list[TwoFactorAction] | None = None,
    remove_2fa: bool = True,
) -> None

Remove 2fa.

Parameters:

Name Type Description Default
password str

Account password.

required
two_factor_actions list[TwoFactorAction] | None

Collection of two factor actions.

None
remove_2fa bool

The remove 2fa value.

True
Source code in src/pyromax/core/CoreMixins/Auth.py
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
async def remove_2fa(
    self,
    password: str,
    two_factor_actions: list[TwoFactorAction] | None = None,
    remove_2fa: bool = True,
) -> None:
    """Remove 2fa.

    :param password: Account password.
    :type password: str
    :param two_factor_actions: Collection of two factor actions.
    :type two_factor_actions: list[TwoFactorAction] | None
    :param remove_2fa: The remove 2fa value.
    :type remove_2fa: bool
    """
    return cast(
        None,
        await self(
            Remove2FaMethod,
            password=password,
            two_factor_actions=two_factor_actions,
            remove_2fa=remove_2fa,
        ),
    )

change_password async

change_password(
    password_old: str,
    password_new: str,
    hint: str | None = None,
    two_factor_actions: list[TwoFactorAction] | None = None,
) -> None

Change password.

Parameters:

Name Type Description Default
password_old str

The password old value.

required
password_new str

The password new value.

required
hint str | None

Password hint.

None
two_factor_actions list[TwoFactorAction] | None

Collection of two factor actions.

None
Source code in src/pyromax/core/CoreMixins/Auth.py
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
async def change_password(
    self,
    password_old: str,
    password_new: str,
    hint: str | None = None,
    two_factor_actions: list[TwoFactorAction] | None = None,
) -> None:
    """Change password.

    :param password_old: The password old value.
    :type password_old: str
    :param password_new: The password new value.
    :type password_new: str
    :param hint: Password hint.
    :type hint: str | None
    :param two_factor_actions: Collection of two factor actions.
    :type two_factor_actions: list[TwoFactorAction] | None
    """
    return cast(
        None,
        await self(
            ChangePasswordMethod,
            password_old=password_old,
            password_new=password_new,
            hint=hint,
            two_factor_actions=two_factor_actions,
        ),
    )

check_2fa async

check_2fa() -> bool

Check 2fa.

Returns:

Type Description
bool

True when the account has 2FA; otherwise False.

Source code in src/pyromax/core/CoreMixins/Auth.py
105
106
107
108
109
110
111
112
113
114
115
116
async def check_2fa(self) -> bool:
    """Check 2fa.

    :returns: True when the account has 2FA; otherwise False.
    :rtype: bool
    """
    return cast(
        bool,
        await self(
            Check2FaMethod,
        ),
    )

approve_qr_login async

approve_qr_login(qr_link: str) -> None

Approve qr login.

Parameters:

Name Type Description Default
qr_link str

Link to the authorization QR code.

required
Source code in src/pyromax/core/CoreMixins/Auth.py
118
119
120
121
122
123
124
125
126
127
128
129
130
async def approve_qr_login(self, qr_link: str) -> None:
    """Approve qr login.

    :param qr_link: Link to the authorization QR code.
    :type qr_link: str
    """
    return cast(
        None,
        await self(
            ApproveQrLoginMethod,
            qr_link=qr_link,
        ),
    )