Requests

zwylib.Requests

class Requests

A utility class providing static/class methods for interacting with Telegram’s API. Most methods are coroutine functions (async def) and must be await-ed, or scheduled via zwylib.async_manager.run_task from synchronous code (e.g. from a hook).

Methods

send

Requests.send(req: TLObject, callback: Optional[Callable], delay: Optional[int] = None, **kwargs) -> None

Low-level, callback-based way to send a raw TL request. Any additional keyword arguments are set as attributes on req before sending, matching the fields in the corresponding TL schema . Prefer async_send unless you specifically need the callback style.

Parameters

• req (TLObject): The request object to send.
• callback (Optional[(TLObject, TLRPC.TL_error) -> None]): Called with the response (or None) and an error (or None).
• delay (Optional[int], default None): Delay in milliseconds before sending the request.
• **kwargs: Additional fields set on req before it’s sent.

Example

def on_response(response, error):
    print(response, error)
 
zwylib.Requests.send(TLRPC.TL_help_getConfig(), on_response)

async_send

Requests.async_send(req: TLObject, delay: Optional[int] = None, **kwargs) -> TLObject

Coroutine version of send — awaits the response instead of taking a callback, and raises an Exception (with the TL error code/text) if the request fails. This is the primitive most other Requests methods are built on.

Parameters

• req (TLObject): The request object to send.
• delay (Optional[int], default None): Delay in milliseconds before sending the request.
• **kwargs: Additional fields set on req before it’s sent.

Example

async def get_config():
    return await zwylib.Requests.async_send(TLRPC.TL_help_getConfig())

reload_admins

Requests.reload_admins(chat_id: int) -> None

Reloads the list of administrators for a given chat.

Parameters

• chat_id (int): ID of the chat to reload administrators for.

Example

zwylib.Requests.reload_admins(chat_id=-12345)
# Reloads admins for the specified chat

delete_messages

Requests.delete_messages(messages: List[int], peer_id: int, topic_id: Optional[int] = None) -> None

Deletes a list of messages from a peer, optionally within a specific topic.

Parameters

• messages (List[int]): List of message IDs to delete.
• peer_id (int): ID of the peer (chat or user) containing the messages.
• topic_id (Optional[int], default None): ID of the topic, if applicable. If None, no topic is specified.

Example

zwylib.Requests.delete_messages(messages=[67890, 67891], peer_id=-12345, topic_id=100)
# Deletes specified messages from the chat

change_slowmode

Requests.change_slowmode(chat_id: int, seconds=0, delay=0) -> None

Changes the slow mode duration for a chat.

Parameters

• chat_id (int): ID of the chat to modify.
• seconds (int, default 0): Number of seconds for the slow mode delay (0 to disable).
• delay (int, default 0): Delay in milliseconds before sending the request.

Example

async def mute_chat(chat_id: int):
    await zwylib.Requests.change_slowmode(chat_id, seconds=30)

ban

Requests.ban(chat_id: int, peer_id: int, until_date: Optional[int] = None, delay=0) -> None

Bans a user in a chat by setting all permissions to restricted, optionally with an expiration date.

Parameters

• chat_id (int): ID of the chat to ban the user in.
• peer_id (int): ID of the user to ban.
• until_date (Optional[int], default None): Unix timestamp when the ban expires (0 or None for permanent).
• delay (int, default 0): Delay in milliseconds before sending the request.

Example

async def ban_user(chat_id: int, peer_id: int):
    await zwylib.Requests.ban(chat_id, peer_id, until_date=1696118400)

unban

Requests.unban(chat_id: int, target_peer_id: int, delay=0) -> None

Removes a ban from a user in a chat, effectively granting them default permissions.

Parameters

• chat_id (int): ID of the chat to unban the user from.
• target_peer_id (int): ID of the user to unban.
• delay (int, default 0): Delay in milliseconds before sending the request.

Example

async def unban_user(chat_id: int, target_peer_id: int):
    await zwylib.Requests.unban(chat_id, target_peer_id)

get_message

Requests.get_message(peer_id: int, message_id: int) -> Optional[TLRPC.messages_Messages]

Shorthand for get_messages with a single message ID. Note that it still returns the full messages_Messages container (with a .messages list), not a bare message object — check .messages for the result.

Parameters

• peer_id (int): ID of the peer (chat or user) containing the message.
• message_id (int): ID of the message to retrieve.

Example

async def print_message(peer_id: int, message_id: int):
    response = await zwylib.Requests.get_message(peer_id, message_id)
    if response.messages:
        print(response.messages[0].message)

get_messages

Requests.get_messages(peer_id: int, messages: List[int]) -> Optional[TLRPC.messages_Messages]

Fetches a batch of messages by ID from a peer, automatically choosing between channels.getMessages and messages.getMessages depending on whether peer_id is a channel.

Parameters

• peer_id (int): ID of the peer (chat or user) containing the messages.
• messages (List[int]): List of message IDs to retrieve.

Example

async def fetch(peer_id: int):
    response = await zwylib.Requests.get_messages(peer_id, [67890, 67891])
    print(f"Got {len(response.messages)} messages")

search_messages

Requests.search_messages(peer_id: int, delay=0, **kwargs) -> TLRPC.messages_Messages

Searches for messages in a peer based on the specified criteria. Additional parameters (e.g., q, offset_id, add_offset, max_id, min_id, min_date, max_date, limit) should be passed as keyword arguments matching the TL schema .

Parameters

• peer_id (int): ID of the peer to search in.
• delay (int, default 0): Delay in milliseconds before sending the request.
• filter (TLRPC.TL_inputMessagesFilter, keyword-only, default TLRPC.TL_inputMessagesFilterEmpty): Filter for message types.
• from_id (Optional[int], keyword-only): ID of the sender to filter messages by.
• top_msg_id (Optional[int], keyword-only): ID of the top message for topic-based search.
• saved_peer_id (Optional[int], keyword-only): ID of the saved messages peer.
• saved_reaction (Optional[ReactionEmoji], keyword-only): Reaction to filter messages by.
• **kwargs: Additional parameters matching the TL schema.

Example

async def search(peer_id: int):
    result = await zwylib.Requests.search_messages(peer_id, q="hello", limit=50)
    print(f"Found {len(result.messages)} messages")

get_chat_participant

Requests.get_chat_participant(chat_id: int, target_peer_id: int) -> TLRPC.TL_channels_channelParticipant

Fetches information about a specific participant in a chat.

Parameters

• chat_id (int): ID of the chat to fetch the participant from.
• target_peer_id (int): ID of the participant to fetch.

Example

async def print_participant(chat_id: int, target_peer_id: int):
    participant = await zwylib.Requests.get_chat_participant(chat_id, target_peer_id)
    print(participant.participant)

Helper Functions

zwylib.process_updates

zwylib.process_updates(updates: TLRPC.TL_updates) -> None

Feeds a TL_updates object (e.g. the result of a request) back into Telegram’s own updates pipeline, so the app’s local state (dialogs, messages, etc.) reflects the change immediately. If updates isn’t a TL_updates instance, it’s ignored. Most Requests methods that return updates already call this for you.

Example

async def kick_and_sync(chat_id: int, peer_id: int):
    ...  # low-level TL_channels_editBanned call
    zwylib.process_updates(response)