Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class SendDocument(BaseResponse, ReplyToMixin, DisableNotificationMixin):
"""
Use that response type for send document on to webhook.
"""
__slots__ = (
"chat_id",
"document",
"caption",
"disable_notification",
"reply_to_message_id",
"reply_markup",
)
method = api.Methods.SEND_DOCUMENT
def __init__(
self,
chat_id: Union[Integer, String],
document: String,
caption: Optional[String] = None,
disable_notification: Optional[Boolean] = None,
reply_to_message_id: Optional[Integer] = None,
reply_markup: Optional[
Union[types.InlineKeyboardMarkup, types.ReplyKeyboardMarkup, Dict, String]
] = None,
):
"""
:param chat_id: Union[Integer, String] - Unique identifier for the target chat or username
of the target channel (in the format @channelusername)
:param document: String - File to send. Pass a file_id as String
class RestrictChatMember(BaseResponse):
"""
Use that response type for restrict chat member on to webhook.
"""
__slots__ = (
"chat_id",
"user_id",
"until_date",
"can_send_messages",
"can_send_media_messages",
"can_send_other_messages",
"can_add_web_page_previews",
)
method = api.Methods.RESTRICT_CHAT_MEMBER
def __init__(
self,
chat_id: Union[Integer, String],
user_id: Integer,
until_date: Optional[Union[Integer, datetime.datetime, datetime.timedelta]] = None,
can_send_messages: Optional[Boolean] = None,
can_send_media_messages: Optional[Boolean] = None,
can_send_other_messages: Optional[Boolean] = None,
can_add_web_page_previews: Optional[Boolean] = None,
):
"""
:param chat_id: Union[Integer, String] - Unique identifier for the target chat
or username of the target supergroup (in the format @supergroupusername)
:param user_id: Integer - Unique identifier of the target user
:param until_date: Integer - Date when restrictions will be lifted for the user, unix time.
class EditMessageText(BaseResponse, ParseModeMixin, DisableWebPagePreviewMixin):
"""
Use that response type for edit message text on to webhook.
"""
__slots__ = (
"chat_id",
"message_id",
"inline_message_id",
"text",
"parse_mode",
"disable_web_page_preview",
"reply_markup",
)
method = api.Methods.EDIT_MESSAGE_TEXT
def __init__(
self,
text: String,
chat_id: Optional[Union[Integer, String]] = None,
message_id: Optional[Integer] = None,
inline_message_id: Optional[String] = None,
parse_mode: Optional[String] = None,
disable_web_page_preview: Optional[Boolean] = None,
reply_markup: Optional[types.InlineKeyboardMarkup] = None,
):
"""
:param chat_id: Union[Integer, String] (Optional) - Required if inline_message_id
is not specified. Unique identifier for the target chat or username of the target channel
(in the format @channelusername)
:param message_id: Integer (Optional) - Required if inline_message_id is not specified.
"""
self.chat_id = chat_id
self.action = action
def prepare(self):
return {"chat_id": self.chat_id, "action": self.action}
class KickChatMember(BaseResponse):
"""
Use that response type for kick chat member on to webhook.
"""
__slots__ = ("chat_id", "user_id", "until_date")
method = api.Methods.KICK_CHAT_MEMBER
def __init__(
self,
chat_id: Union[Integer, String],
user_id: Integer,
until_date: Optional[Union[Integer, datetime.datetime, datetime.timedelta]] = None,
):
"""
:param chat_id: Union[Integer, String] - Unique identifier for the target group or username
of the target supergroup or channel (in the format @channelusername)
:param user_id: Integer - Unique identifier of the target user
:param until_date: Integer - Date when the user will be unbanned, unix time. If user is banned for
more than 366 days or less than 30 seconds from the current time they are considered to be banned forever
"""
self.chat_id = chat_id
self.user_id = user_id
return {
"chat_id": self.chat_id,
"message_id": self.message_id,
"inline_message_id": self.inline_message_id,
"reply_markup": prepare_arg(self.reply_markup),
}
class DeleteMessage(BaseResponse):
"""
Use that response type for delete message on to webhook.
"""
__slots__ = ("chat_id", "message_id")
method = api.Methods.DELETE_MESSAGE
def __init__(self, chat_id: Union[Integer, String], message_id: Integer):
"""
:param chat_id: Union[Integer, String] - Unique identifier for the target chat or username
of the target channel (in the format @channelusername)
:param message_id: Integer - Identifier of the message to delete
"""
self.chat_id = chat_id
self.message_id = message_id
def prepare(self):
return {"chat_id": self.chat_id, "message_id": self.message_id}
class SendSticker(BaseResponse, ReplyToMixin, DisableNotificationMixin):
"""
"title": self.title,
"png_sticker": self.png_sticker,
"emojis": self.emojis,
"contains_masks": self.contains_masks,
"mask_position": self.mask_position,
}
class AddStickerToSet(BaseResponse):
"""
Use that response type for add sticker to set on to webhook.
"""
__slots__ = ("user_id", "name", "png_sticker", "emojis", "mask_position")
method = api.Methods.ADD_STICKER_TO_SET
def __init__(
self,
user_id: Integer,
name: String,
png_sticker: String,
emojis: String,
mask_position: Optional[types.MaskPosition] = None,
):
"""
:param user_id: Integer - User identifier of sticker set owner
:param name: String - Sticker set name
:param png_sticker: String - Png image with the sticker,
must be up to 512 kilobytes in size, dimensions must not exceed 512px,
and either width or height must be exactly 512px. Pass a file_id as a String
to send a file that already exists on the Telegram servers, pass an HTTP URL
Use that response type for send audio on to webhook.
"""
__slots__ = (
"chat_id",
"audio",
"caption",
"duration",
"performer",
"title",
"disable_notification",
"reply_to_message_id",
"reply_markup",
)
method = api.Methods.SEND_AUDIO
def __init__(
self,
chat_id: Union[Integer, String],
audio: String,
caption: Optional[String] = None,
duration: Optional[Integer] = None,
performer: Optional[String] = None,
title: Optional[String] = None,
disable_notification: Optional[Boolean] = None,
reply_to_message_id: Optional[Integer] = None,
reply_markup: Optional[
Union[types.InlineKeyboardMarkup, types.ReplyKeyboardMarkup, Dict, String]
] = None,
):
"""
class AnswerInlineQuery(BaseResponse):
"""
Use that response type for answer inline query on to webhook.
"""
__slots__ = (
"inline_query_id",
"results",
"cache_time",
"is_personal",
"next_offset",
"switch_pm_text",
"switch_pm_parameter",
)
method = api.Methods.ANSWER_INLINE_QUERY
def __init__(
self,
inline_query_id: String,
results: [types.InlineQueryResult],
cache_time: Optional[Integer] = None,
is_personal: Optional[Boolean] = None,
next_offset: Optional[String] = None,
switch_pm_text: Optional[String] = None,
switch_pm_parameter: Optional[String] = None,
):
"""
:param inline_query_id: String - Unique identifier for the answered query
:param results: [types.InlineQueryResult] - A JSON-serialized array of results for the inline query
:param cache_time: Integer (Optional) - The maximum amount of time in seconds that the result
of the inline query may be cached on the server. Defaults to 300.
"""
You can send message with webhook by using this instance of this object.
All arguments is equal with Bot.send_message method.
"""
__slots__ = (
"chat_id",
"text",
"parse_mode",
"disable_web_page_preview",
"disable_notification",
"reply_to_message_id",
"reply_markup",
)
method = api.Methods.SEND_MESSAGE
def __init__(
self,
chat_id: Union[Integer, String] = None,
text: String = None,
parse_mode: Optional[String] = None,
disable_web_page_preview: Optional[Boolean] = None,
disable_notification: Optional[Boolean] = None,
reply_to_message_id: Optional[Integer] = None,
reply_markup: Optional[
Union[types.InlineKeyboardMarkup, types.ReplyKeyboardMarkup, Dict, String]
] = None,
):
"""
:param chat_id: Union[Integer, String] - Unique identifier for the target chat or username
of the target channel (in the format @channelusername)
async def make_request(session, token, method, data=None, files=None, **kwargs):
# log.debug(f"Make request: '{method}' with data: {data} and files {files}")
log.debug('Make request: "%s" with data: "%r" and files "%r"', method, data, files)
url = Methods.api_url(token=token, method=method)
req = compose_data(data, files)
try:
async with session.post(url, data=req, **kwargs) as response:
return check_result(method, response.content_type, response.status, await response.text())
except aiohttp.ClientError as e:
raise exceptions.NetworkError(f"aiohttp client throws an error: {e.__class__.__name__}: {e}")