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	DeviceType | None	Device type reported to the API.	None
password	str | None	Optional account password.	None
auth_callbacks	AuthCallbacks | None	Callbacks for QR, SMS code, and 2FA password.	None
auth_interactor	AuthInteractor | None	Custom authentication interaction implementation.	None
token	str | None	Optional auth token.	None
transport	BaseTransport[Any] | 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	The token suffix value.	'MaxApi'

Raises:

Type	Description
RuntimeError	If transport or protocol or mapper cannot be None.

Source code in src/pyromax/core/client.py

def __init__(
    self,
    device_type: DeviceType | None = None,  # default "WEB",
    token_suffix: str = "MaxApi",
    password: str | None = None,
    auth_callbacks: AuthCallbacks | None = None,
    auth_interactor: AuthInteractor | None = None,
    session_id: str | None = None,
    transport: BaseTransport[Any] | None = None,
    encoding: BaseEncoding[Any, Any, Any, Any] | 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,
    extra_config: ExtraConfig | 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 auth_callbacks: Callbacks for QR, SMS code, and 2FA password.
    :param auth_interactor: Custom authentication interaction implementation.
    :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.
    """
    self.name: str
    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")

    if extra_config is None:
        raise RuntimeError("extra_config cannot be None")
    self.extra_config = extra_config

    self.transport = transport
    self.transport_options = transport_options
    self.protocol = protocol
    self.mapper = mapper
    self.id: int | 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 = logger
    self._connected = False
    self.workflow_data = workflow_data
    self.auth_middleware_manager = auth_middleware_manager

_SHUTDOWN_STEP_TIMEOUT class-attribute instance-attribute

_SHUTDOWN_STEP_TIMEOUT = 5.0

device_type instance-attribute

device_type: DeviceType

session instance-attribute

session: SessionInfo

shutdown_requested property

shutdown_requested: bool

Return whether graceful shutdown has been requested.

name instance-attribute

name: str

extra_config instance-attribute

extra_config = extra_config

transport instance-attribute

transport = transport

transport_options instance-attribute

transport_options = transport_options

protocol instance-attribute

protocol = protocol

mapper instance-attribute

mapper = mapper

id instance-attribute

id: int | 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 = logger

_connected instance-attribute

_connected = False

workflow_data instance-attribute

workflow_data = workflow_data

auth_middleware_manager instance-attribute

auth_middleware_manager: AuthMiddlewareManager | None = (
    auth_middleware_manager
)

session_key property

session_key: SessionKey

phone property writable

phone: str | None

password property writable

password: str | None

token property writable

token: str | None

session_id property writable

session_id: str | None

_async_init async

_async_init(
    device_type: DeviceType | None = None,
    password: str | None = None,
    auth_callbacks: AuthCallbacks | None = None,
    auth_interactor: AuthInteractor | None = None,
    token: str | None = None,
    session_id: str | None = None,
    phone: str | None = None,
    token_suffix: str = "MaxApi",
    work_dir: str | None = None,
    session_name: str | None = None,
    transport: TransportRegistry | None = None,
    encoding: EncodingRegistry | None = None,
    protocol: ProtocolRegistry | None = None,
    mapper: MapperRegistry | None = None,
    workflow_data: dict[Any, Any] | None = None,
    auth_middleware_manager: (
        AuthMiddlewareManager | None
    ) = None,
    extra_config: ExtraConfig | None = None,
    **kwargs: Any
) -> None

Asynchronously initialize transport, protocol, and mapper.

Parameters:

Name	Type	Description	Default
device_type	DeviceType | None	Device type reported to the API.	None
password	str | None	Optional account password.	None
auth_callbacks	AuthCallbacks | None	Callbacks for QR, SMS code, and 2FA password.	None
auth_interactor	AuthInteractor | None	Custom authentication interaction implementation.	None
token	str | None	Optional auth token.	None
transport	TransportRegistry | None	Transport backend name.	None
protocol	ProtocolRegistry | None	Protocol backend name.	None
mapper	MapperRegistry | None	Mapper backend name.	None
transport_options	dict[str, Any] | None	Keyword arguments passed to the transport constructor.	required
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.	required
token_suffix	str	The token suffix value.	'MaxApi'

Raises:

Type	Description
RuntimeError	If transport or protocol or mapper cannot be None.

Source code in src/pyromax/core/client.py

async def _async_init(
    self,
    device_type: DeviceType | None = None,  # default "WEB"
    password: str | None = None,
    auth_callbacks: AuthCallbacks | None = None,
    auth_interactor: AuthInteractor | None = None,
    token: str | None = None,
    session_id: str | None = None,
    phone: str | None = None,
    token_suffix: str = "MaxApi",
    work_dir: str | None = None,
    session_name: str | None = None,
    transport: TransportRegistry | None = None,
    encoding: EncodingRegistry | None = None,
    protocol: ProtocolRegistry | None = None,
    mapper: MapperRegistry | None = None,
    workflow_data: dict[Any, Any] | None = None,
    auth_middleware_manager: AuthMiddlewareManager | None = None,
    extra_config: ExtraConfig | 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.
    :param auth_callbacks: Callbacks for QR, SMS code, and 2FA password.
    :param auth_interactor: Custom authentication interaction implementation.
    :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.
    """

    self._shutdown_event = asyncio.Event()
    self._shutdown_worker_task: asyncio.Task[None] | None = None
    self._stop_task: asyncio.Task[None] | None = None
    self._initialization_task: asyncio.Task[Any] | None = asyncio.current_task()
    self._logger = logging.getLogger("MaxApi")
    self.transport = cast("BaseTransport[Any]", None)
    self.protocol = cast("BaseMaxProtocol[Any, Any]", None)
    self.mapper = cast("BaseMapper[Any, Any]", None)
    self._session_updates_task: asyncio.Task[None] | None = None
    self.session_storage = cast("BaseSessionStorage", None)

    from ..models import (
        DeviceType,
        TransportRegistry,
        EncodingRegistry,
        ProtocolRegistry,
        MapperRegistry,
        SessionInfo,
    )

    if extra_config is None:
        device_type = device_type or DeviceType.Web

        if device_type is None:
            # dummy for type-checker, because it see it by "str | DeviceType | None | Literal[DeviceType.Web]",
            # not "str | DeviceType"
            raise RuntimeError("Never")
    else:
        device_type = (
            device_type or extra_config.mapper.device_type or DeviceType.Web
        )

        if device_type is None:
            # dummy for type-checker, because it see it by "str | DeviceType | None | Literal[DeviceType.Web]",
            # not "str | DeviceType"
            raise RuntimeError("Never")

        extra_config.mapper.device_type = device_type
    self.device_type = device_type

    if extra_config is None:
        default_mapper_conf = type(ExtraConfig().mapper)
        extra_config = ExtraConfig(
            mapper=default_mapper_conf(
                token=token,
                password=password,
                device_type=device_type,
                phone=phone,
            ),
            auth_interactor=auth_interactor,
        )

    if token is not None:
        extra_config.mapper.token = token
    if password is not None:
        extra_config.mapper.password = password
    if phone is not None:
        extra_config.mapper.phone = phone
    if auth_interactor is not None:
        extra_config.auth_interactor = auth_interactor
    if auth_callbacks is not None and extra_config.auth_interactor is not None:
        raise ValueError(
            "auth_callbacks and a custom auth_interactor cannot be used together"
        )
    if extra_config.auth_interactor is None:
        from ..interaction import TerminalAuthInteractor

        extra_config.auth_interactor = TerminalAuthInteractor(auth_callbacks)
    if session_id is not None:
        extra_config.session_id = session_id
    if work_dir is not None:
        extra_config.work_dir = work_dir
    if session_name is not None:
        extra_config.session_name = session_name

    self.extra_config = extra_config.config_rebuild(
        transport=transport,
        protocol=protocol,
        encoding=encoding,
        mapper=mapper,
    )

    self.session = SessionInfo(
        phone=self.extra_config.mapper.phone,
        device_id=self.extra_config.mapper.user_agent_config.device_id,
        token=self.extra_config.mapper.token,
        session_id=self.extra_config.session_id,
        user_agent_config=self.extra_config.mapper.user_agent_config.to_string(),
        **self.extra_config.mapper.user_agent_config.session_info_params,
    )
    self.token_suffix = token_suffix
    try:
        self._session_updates_queue: asyncio.Queue[tuple[SessionKey, SessionInfo]] = (
            asyncio.Queue()
        )
        self.session_storage = (
            self.extra_config.session_storage
            or AioSqLiteSessionStorage(
                work_dir=self.extra_config.work_dir,
                db_name=self.extra_config.session_name,
            )
        )
        self._session_updates_task = asyncio.create_task(self._update_session())
        self._shutdown_worker_task = asyncio.create_task(self._shutdown_worker())

        session_info = await self.session_storage.load_session(self.session_key)

        if session_info is not None:
            if self.session_id is None and session_info.session_id is not None:
                self.session_id = session_info.session_id
            elif self.session_id is None and session_info.session_id is None:
                self.session_id = session_info.session_id = str(uuid.uuid4())

            if self.token is None:
                self.token = session_info.token
            if self.phone is None:
                self.phone = session_info.phone
            custom_user_agent_config = (
                extra_config.mapper.is_custom_user_agent_config
            )
            if (
                session_info.user_agent_config is not None
                and not custom_user_agent_config
                and self.extra_config.restore_user_agent_from_session
            ):
                user_agent_config = (
                    BaseEnvelopeMappingUserAgentConfigV11.from_session_info(
                        session_info
                    )
                )
                self.extra_config.mapper.user_agent_config = user_agent_config
                self.session.user_agent_config = session_info.user_agent_config
            if session_info.sync:
                current_sync = self.session.sync
                saved_sync = session_info.sync

                from ..models.Session import SyncState

                default_sync = SyncState()

                if current_sync.presence_sync == default_sync.presence_sync:
                    current_sync.presence_sync = saved_sync.presence_sync
                if current_sync.chats_sync == default_sync.chats_sync:
                    current_sync.chats_sync = saved_sync.chats_sync
                if current_sync.drafts_sync == default_sync.drafts_sync:
                    current_sync.drafts_sync = saved_sync.drafts_sync
                if current_sync.contacts_sync == default_sync.contacts_sync:
                    current_sync.contacts_sync = saved_sync.contacts_sync
                if str(current_sync.config_hash) == str(default_sync.config_hash):
                    current_sync.config_hash = saved_sync.config_hash
        else:
            if self.session_id is None:
                self.session_id = str(uuid.uuid4())

        transport = cast(
            TransportRegistry,
            from_config_to_registry(type(self.extra_config.transport)),
        )
        encoding = cast(
            EncodingRegistry,
            from_config_to_registry(type(self.extra_config.encoding)),
        )
        protocol = cast(
            ProtocolRegistry,
            from_config_to_registry(type(self.extra_config.protocol)),
        )
        mapper = cast(
            MapperRegistry, from_config_to_registry(type(self.extra_config.mapper))
        )

        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...")

        max_encoding: BaseEncoding[Any, Any, Any, Any] = from_registry(
            ENCODINGS, encoding
        )(self.extra_config)

        logger.info("Initializing transport...")

        max_transport = await from_registry(TRANSPORTS, transport)(
            max_encoding, self.extra_config
        )
        self.transport = max_transport
        logger.info("Transport initialized.")

        logger.info("Initializing protocol...")

        protocol_res: Any = await from_registry(PROTOCOLS, protocol)(
            transport=max_transport,
            encoding=max_encoding,
            extra_config=self.extra_config,
        )
        max_protocol: BaseMaxProtocol[Any, Any] = protocol_res
        self.protocol = max_protocol
        logger.info("Protocol initialized.")

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

        hide_func_call(
            type(self).__init__,
            self,
            protocol=max_protocol,
            password=extra_config.mapper.password,
            transport=max_transport,
            mapper=max_mapper,
            token=extra_config.mapper.token,
            logger=logger,
            workflow_data=workflow_data,
            device_type=device_type,
            auth_middleware_manager=auth_middleware_manager,
            extra_config=self.extra_config,
        )
        if self.extra_config.auto_connect_after_init:
            await self.connect(**kwargs)
    except asyncio.CancelledError:
        try:
            await self.stop()
        except Exception:
            if self._logger is not None:
                self._logger.exception(
                    "Cleanup failed while cancelling client initialization"
                )
        raise
    finally:
        self._initialization_task = None

connect async

connect(**kwargs: Any) -> None

Source code in src/pyromax/core/client.py

async def connect(
    self,
    **kwargs: Any,
) -> None:
    if self.shutdown_requested:
        stop_task = self._stop_task
        if stop_task is not None and not stop_task.done():
            await asyncio.shield(stop_task)
        self._shutdown_event.clear()
        self._stop_task = None
        self._shutdown_worker_task = asyncio.create_task(self._shutdown_worker())

    if self._session_updates_task is None:
        self._session_updates_task = asyncio.create_task(self._update_session())

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

        await self.mapper.start_auth_flow(
            max_api=self,
            extra_config=self.extra_config,
            **kwargs,
        )

        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

        if token:
            await self.mapper.end_auth_flow(token)
            self.token = token
        else:
            await self.mapper.end_auth_flow(None)

    await self.mapper.start()
    self._connected = True

request_shutdown

request_shutdown() -> None

Request graceful shutdown without blocking the caller.

Source code in src/pyromax/core/client.py

def request_shutdown(self) -> None:
    """Request graceful shutdown without blocking the caller."""
    self._shutdown_event.set()
    initialization_task = self._initialization_task
    if (
        initialization_task is not None
        and not initialization_task.done()
        and not initialization_task.cancelling()
    ):
        initialization_task.cancel()

wait_for_shutdown async

wait_for_shutdown() -> None

Wait until graceful shutdown is requested.

Source code in src/pyromax/core/client.py

async def wait_for_shutdown(self) -> None:
    """Wait until graceful shutdown is requested."""
    await self._shutdown_event.wait()

_shutdown_worker async

_shutdown_worker() -> None

Source code in src/pyromax/core/client.py

async def _shutdown_worker(self) -> None:
    try:
        await self._shutdown_event.wait()
        await self.stop()
    except asyncio.CancelledError:
        raise
    except Exception:
        logger = getattr(self, "_logger", None)
        if logger is not None:
            logger.exception("Graceful shutdown failed")

_stop async

_stop() -> None

Source code in src/pyromax/core/client.py

async def _stop(self) -> None:
    try:
        self._connected = False
        mapper = getattr(self, "mapper", None)
        if mapper is not None:
            try:
                await asyncio.wait_for(
                    mapper.stop(), timeout=self._SHUTDOWN_STEP_TIMEOUT
                )
            except asyncio.TimeoutError:
                if self._logger is not None:
                    self._logger.warning("Mapper shutdown timed out")
        else:
            protocol = getattr(self, "protocol", None)
            transport = getattr(self, "transport", None)
            if protocol is not None:
                try:
                    await asyncio.wait_for(
                        protocol.close(), timeout=self._SHUTDOWN_STEP_TIMEOUT
                    )
                except asyncio.TimeoutError:
                    if self._logger is not None:
                        self._logger.warning("Protocol shutdown timed out")
            elif transport is not None:
                try:
                    await asyncio.wait_for(
                        transport.close(), timeout=self._SHUTDOWN_STEP_TIMEOUT
                    )
                except asyncio.TimeoutError:
                    if self._logger is not None:
                        self._logger.warning("Transport shutdown timed out")

        session_updates_task = getattr(self, "_session_updates_task", None)
        if session_updates_task is not None:
            try:
                await asyncio.wait_for(
                    self._session_updates_queue.join(),
                    timeout=self._SHUTDOWN_STEP_TIMEOUT,
                )
            except asyncio.TimeoutError:
                if self._logger is not None:
                    self._logger.warning("Session updates drain timed out")
    finally:
        updates_task = getattr(self, "_session_updates_task", None)
        self._session_updates_task = None
        if updates_task is not None:
            updates_task.cancel()
            with suppress(asyncio.CancelledError):
                await updates_task

        session_storage = getattr(self, "session_storage", None)
        if session_storage is not None:
            try:
                await asyncio.wait_for(
                    session_storage.close(),
                    timeout=self._SHUTDOWN_STEP_TIMEOUT,
                )
            except asyncio.TimeoutError:
                if self._logger is not None:
                    self._logger.warning("Session storage shutdown timed out")

stop async

stop() -> None

Stop the client once and wait for graceful shutdown to finish.

Source code in src/pyromax/core/client.py

async def stop(self) -> None:
    """Stop the client once and wait for graceful shutdown to finish."""
    self.request_shutdown()
    if self._stop_task is None:
        self._stop_task = asyncio.create_task(self._stop())
    await asyncio.shield(self._stop_task)

wait_until_ready async

wait_until_ready() -> bool

Wait until the client is ready or shutdown is requested.

Source code in src/pyromax/core/client.py

async def wait_until_ready(self) -> bool:
    """Wait until the client is ready or shutdown is requested."""
    if self.shutdown_requested:
        return False

    ready_task = asyncio.create_task(self.mapper.wait_until_ready())
    shutdown_task = asyncio.create_task(self.wait_for_shutdown())
    try:
        await asyncio.wait(
            (ready_task, shutdown_task),
            return_when=asyncio.FIRST_COMPLETED,
        )
        return ready_task.done() and not self.shutdown_requested
    finally:
        for task in (ready_task, shutdown_task):
            if not task.done():
                task.cancel()
        await asyncio.gather(ready_task, shutdown_task, return_exceptions=True)

_update_session async

_update_session() -> None

Source code in src/pyromax/core/client.py

async def _update_session(self) -> None:

    while True:
        session_key, session_update = await self._session_updates_queue.get()
        try:
            await self.session_storage.update_session(session_key, session_update)
        except Exception as e:
            self._logger.error(
                "Error while updating session=%s, by session_key=%s",
                session_update,
                session_key,
                exc_info=e,
            )
        finally:
            self._session_updates_queue.task_done()

__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 without connect state, because logger has not been initialized.

Source code in src/pyromax/core/client.py

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 without connect state, because logger has not been initialized.
    """
    if self._logger is None:
        raise RuntimeError(
            "Try a call method before initialize, because logger has not been initialized"
        )

    if not self._connected:
        raise RuntimeError("Try a call method without connect state")
    self._logger.debug("Calling MaxApi method: %s", class_of_method.__name__)
    method = class_of_method().as_(self)
    try:
        return await method(*args, **kwargs)
    except MapperTransportError as e:
        self._logger.error(
            "Mapper transport error while call method=%s: %s",
            class_of_method.__name__,
            e,
        )
        raise BaseMaxApiMethodError(
            "error while call method=%s: %s",
            class_of_method.__name__,
            e,
        ) from e

    except MapperApiError as e:
        self._logger.error(
            "Mapper API error while call method=%s: %s",
            class_of_method.__name__,
            e,
        )
        raise BaseMaxApiMethodError(
            "API error title=%s error=%s message=%s localized_message=%s",
            e.title,
            e.error,
            e.message,
            e.localized_message,
        ) from e

    except BaseMapperError as e:
        self._logger.error(
            "Mapper unknown error while call method=%s: %s",
            class_of_method.__name__,
            e,
        )
        raise BaseMaxApiMethodError(
            "error while call method=%s: %s",
            class_of_method.__name__,
            e,
        ) from e

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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 async

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

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 async

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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
    """
    msg_ids: list[int] | list[str]
    if isinstance(message_id, str):
        msg_ids = [message_id]
    else:
        msg_ids = [message_id]

    msgs = await self.get_messages(chat_id=chat_id, message_ids=msg_ids)
    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: Literal[False] = False,
    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: Literal[True] = True,
    get_messages: Literal[True] = True,
    interactive: bool = ...,
) -> tuple[list[Message], Chat | None]

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: Literal[False] = False,
    get_messages: Literal[False] = False,
    interactive: bool = ...,
) -> list[str | int]

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: Literal[True] = True,
    get_messages: Literal[False] = False,
    interactive: bool = ...,
) -> tuple[list[str | int], Chat | None]

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 | int]
    | tuple[list[Message] | list[str | int], Chat | None]
)

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 | int]
    | tuple[list[Message] | list[str | int], Chat | None]
)

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 | int] | tuple[list[Message] | list[str | int], Chat | None]	History items, paired with the requested chat when get_chat is true.

Source code in src/pyromax/core/CoreMixins/Message.py

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 | int]
    | tuple[list[Message] | list[str | int], Chat | None]
):
    """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: History items, paired with the requested chat when ``get_chat`` is true.
    :rtype: list[Message] | list[str | int] | tuple[list[Message] | list[str | int], Chat | None]
    """

    return cast(
        list[Message]
        | list[str | int]
        | tuple[list[Message] | list[str | int], Chat | None],
        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

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

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

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

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

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

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

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

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

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

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

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

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

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,
        ),
    )