Quickstart
Как установить
Библиотека доступна на PyPI:
`shell
$ python -m pip install tg-api
`
Поддерживаемые версии Python и зависимостей
Поддерживаются:
Ключевые концепции
Библиотека Tg API предлагает несколько необычных концепций для работы с API. Пробежимся по ним вкратце.
No God Object
Библиотека Tg API не предоставляет пользователю никакого
аналога “god object” для работы с API, как то TgBot
или TgApi
. В
других библиотеках часто можно увидеть подобный код:
bot = TgBot(token=...)
bot.send_message(text='Hello world!', chat_id=43)
Такой подход прекрасно выглядит в туториалах, он кажется простым и естественным, но ему сильно не хватает гибкости. При интенсивном использовании и кастомизации вы неизбежно столкнётесь с нехваткой документации, неожиданными ограничениями ПО и вам придётся лезть в код библиотеки, чтобы решить свою проблему. Подробно типичные проблемы такого подхода описаны в антипаттерне God object.
В библиотеке Tg API нет и не будет аналога объекта Bot
. Вместо него
для отправки запросов используются объекты SendMessageRequest
,
SendPhotoRequest
и подобные Request-объекты, по одному для каждого
метода Tg Bot API из документации
Telegram. Сначала вы
готовите запрос к API, затем отправляете и обрабатываете результат.
Пример:
# создаём объект запроса, но ещё не отправляем
tg_request = SendMessageRequest(text='Hello world!', chat_id=43)
# отправляем запрос в API
# вызов метода поднимет исключение TgRuntimeError если сервер Telegram ответит
# на запрос HTTP статусом != 2xx
tg_response: SendMessageResponse = tg_request.send()
Преимущество такого подхода в том, что он не создаёт лишних обёрток над
схемой запроса и ответа к API. Вам не нужно искать документацию по
методу send_message
, не нужно мириться с ограничениями этого метода.
Вы сможете отправлять в API даже запросы с крайне нетипичными
параметрами, и полная схема доступных параметров у вас всегда под рукой.
Default configuration
Вам не нужен прямой доступ к объекту
TgBot
, TgApi
или TgClient
для работы с API. Обычно,
приходится таскать подобный объект за собой из функции в функцию, чтобы
где-то там глубоко внутри отправить пользователю сообщение в Tg.
Библиотека Tg API
использует contextvars
, чтобы передавать
настройки подключения неявно. Пример:
def do_something():
# Function send message without direct access to TgClient object
tg_request = SendMessageRequest(text='Hello world!', chat_id=43)
tg_request.send()
def main(token: str) -> None:
with TgClient.setup(token):
do_something()
Синхронные примеры использования
Пример отправки пользователю текстового сообщения:
from tg_api import SyncTgClient, SendMessageRequest
with SyncTgClient.setup(token):
tg_request = SendMessageRequest(chat_id=tg_chat_id, text='Hello user!')
tg_request.send()
Пример удаления у пользователя любого сообщения по идентификатору сообщения:
from tg_api import SyncTgClient, DeleteMessageRequest
with SyncTgClient.setup(token):
tg_request = DeleteMessageRequest(chat_id=tg_chat_id, message_id=message_id)
tg_request.send()
Пример изменения у пользователя текста любого сообщения по идентификатору сообщения:
from tg_api import SyncTgClient, EditMessageTextRequest
with SyncTgClient.setup(token):
tg_request = EditMessageTextRequest(
chat_id=tg_chat_id,
message_id=message_id,
text='edited text',
)
tg_request.send()
Пример изменения у пользователя заголовка сообщения по идентификатору сообщения:
from tg_api import SyncTgClient, EditMessageCaptionRequest
with SyncTgClient.setup(token):
tg_request = EditMessageCaptionRequest(
chat_id=chat_id,
message_id=message_id,
caption='edited caption',
)
tg_request.send()
Пример изменения у пользователя фото в сообщении по URL по идентификатору сообщения:
from tg_api import SyncTgClient, EditUrlMessageMediaRequest
with SyncTgClient.setup(token):
media = InputMediaUrlDocument(
media='https://link_to_photo.jpg',
caption='caption'
)
tg_request = EditUrlMessageMediaRequest(
chat_id=chat_id,
message_id=message_id,
media=media,
)
tg_request.send()
Пример изменения у пользователя документа в сообщении чтением документа из файла по идентификатору сообщения:
from tg_api import SyncTgClient, EditBytesMessageMediaRequest, InputMediaBytesDocument
with SyncTgClient.setup(token):
with open('path_to_document.pdf', 'rb') as f:
media_content = f.read()
media = InputMediaBytesDocument(
media='attach://attachement.pdf',
media_content=media_content,
caption='caption'
)
tg_request = EditBytesMessageMediaRequest(
chat_id=chat_id,
message_id=message_id,
media=media,
)
tg_request.send()
Пример изменения у пользователя клавиатуры любого сообщения по идентификатору сообщения:
from tg_api import SyncTgClient, InlineKeyboardButton, InlineKeyboardMarkup
keyboard = InlineKeyboardMarkup(
inline_keyboard=[
[
InlineKeyboardButton(text='button_1', callback_data='test'),
InlineKeyboardButton(text='button_2', callback_data='test'),
],
],
)
with SyncTgClient.setup(token):
tg_request = EditMessageReplyMarkupRequest(
chat_id=tg_chat_id,
message_id=message_id,
reply_markup=keyboard,
)
tg_request.send()
Пример отправки пользователю сообщения с клавиатурой:
from tg_api import (
SyncTgClient,
SendMessageRequest,
InlineKeyboardButton,
InlineKeyboardMarkup,
)
keyboard = InlineKeyboardMarkup(
inline_keyboard=[
[
InlineKeyboardButton(text='button_1', callback_data='test'),
InlineKeyboardButton(text='button_2', callback_data='test'),
],
],
)
with SyncTgClient.setup(token):
tg_request = SendMessageRequest(
chat_id=tg_chat_id,
text='Message proofs keyboard support.',
reply_markup=keyboard,
)
tg_request.send()
Пример отправки пользователю фото из файловой системы:
from tg_api import SyncTgClient, SendBytesPhotoRequest
with SyncTgClient.setup(token):
with open(photo_filename, 'rb') as f:
photo_content = f.read()
tg_request = SendBytesPhotoRequest(
chat_id=chat_id,
photo=photo_content,
filename=photo_filename,
)
tg_request.send()
Пример отправки пользователю фото по URL:
from tg_api import SyncTgClient, SendUrlPhotoRequest
with SyncTgClient.setup(token):
tg_request = SendUrlPhotoRequest(
chat_id=chat_id,
photo=photo_url,
filename=photo_filename,
)
tg_request.send()
Пример отправки пользователю документа из файловой системы:
from tg_api import SyncTgClient, SendBytesDocumentRequest
with SyncTgClient.setup(token):
with open(document_filename, 'rb') as f:
document_content = f.read()
tg_request = SendBytesDocumentRequest(
chat_id=chat_id,
document=document_content,
filename=document_filename,
)
tg_request.send()
Пример отправки пользователю документа по URL:
from tg_api import SyncTgClient, SendUrlDocumentRequest
with SyncTgClient.setup(token):
tg_request = SendUrlDocumentRequest(
chat_id=chat_id,
document=document_url,
filename=document_filename,
)
tg_request.send()
Пример поллинга:
from contextlib import suppress
from tg_api import GetUpdatesRequest, SyncTgClient, Update
TG_BOT_TOKEN = "replace me with a real token"
def handle_update(update: Update) -> None:
print(vars(update)) # your logic comes here
def main() -> None:
with SyncTgClient.setup(TG_BOT_TOKEN):
tg_request = GetUpdatesRequest(timeout=30)
for update in tg_request.listen_to_updates():
handle_update(update)
if __name__ == '__main__':
with suppress(KeyboardInterrupt):
main()
Асинхронные примеры использования
Пример отправки пользователю текстового сообщения:
from tg_api import AsyncTgClient, SendMessageRequest
async with AsyncTgClient.setup(token):
tg_request = SendMessageRequest(
chat_id=chat_id,
text='Message proofs high level API usage.',
)
# вызов метода поднимет исключение TgRuntimeError если сервер Telegram ответит
# на запрос HTTP статусом != 2xx
await tg_request.asend()
Пример удаления у пользователя любого сообщения по идентификатору сообщения:
from tg_api import AsyncTgClient, DeleteMessageRequest
async with AsyncTgClient.setup(token):
tg_request = DeleteMessageRequest(chat_id=chat_id, message_id=message_id)
await tg_request.asend()
Пример изменения у пользователя текста любого сообщения по идентификатору сообщения:
from tg_api import AsyncTgClient, EditMessageTextRequest
async with AsyncTgClient.setup(token):
tg_request = EditMessageTextRequest(
chat_id=chat_id,
message_id=message_id,
text='edited text',
)
await tg_request.asend()
Пример изменения у пользователя фото в сообщении по URL по идентификатору сообщения:
from tg_api import AsyncTgClient, EditUrlMessageMediaRequest
async with AsyncTgClient.setup(token):
media = InputMediaUrlDocument(
media='https://link_to_photo.jpg',
caption='caption'
)
tg_request = EditUrlMessageMediaRequest(
chat_id=chat_id,
message_id=message_id,
media=media,
)
await tg_request.asend()
Пример изменения у пользователя документа в сообщении чтением документа из файла по идентификатору сообщения:
from tg_api import AsyncTgClient, EditBytesMessageMediaRequest, InputMediaBytesDocument
async with AsyncTgClient.setup(token):
with open('path_to_document.pdf', 'rb') as f:
media_content = f.read()
media = InputMediaBytesDocument(
media='attach://attachement.pdf',
media_content=media_content,
caption='caption'
)
tg_request = EditBytesMessageMediaRequest(
chat_id=chat_id,
message_id=message_id,
media=media,
)
await tg_request.asend()
Пример изменения у пользователя клавиатуры любого сообщения по идентификатору сообщения:
from tg_api import AsyncTgClient, InlineKeyboardButton, InlineKeyboardMarkup
keyboard = InlineKeyboardMarkup(
inline_keyboard=[
[
InlineKeyboardButton(text='button_1', callback_data='test'),
InlineKeyboardButton(text='button_2', callback_data='test'),
],
],
)
async with AsyncTgClient.setup(token):
tg_request = EditMessageReplyMarkupRequest(
chat_id=chat_id,
message_id=message_id,
reply_markup=keyboard,
)
await tg_request.asend()
Пример изменения у пользователя заголовка сообщения по идентификатору сообщения:
from tg_api import AsyncTgClient, EditMessageCaptionRequest
async with AsyncTgClient.setup(token):
tg_request = EditMessageCaptionRequest(
chat_id=chat_id,
message_id=message_id,
caption='edited caption',
)
await tg_request.asend()
Пример отправки пользователю сообщения с клавиатурой:
from tg_api import (
AsyncTgClient,
SendMessageRequest,
InlineKeyboardButton,
InlineKeyboardMarkup,
)
async def main(token: str, chat_id: int) -> None:
keyboard = InlineKeyboardMarkup(
inline_keyboard=[
[
InlineKeyboardButton(text='button_1', callback_data='test'),
InlineKeyboardButton(text='button_2', callback_data='test'),
],
],
)
async with AsyncTgClient.setup(token):
tg_request = SendMessageRequest(
chat_id=chat_id,
text='Message proofs keyboard support.',
reply_markup=keyboard,
)
await tg_request.asend()
Пример отправки пользователю фото из файловой системы:
import aiofiles
import tg_api
async def main(token: str, chat_id: int, photo_filename: str) -> None:
async with tg_api.AsyncTgClient.setup(token):
async with aiofiles.open(photo_filename, 'rb') as f:
photo_content = await f.read()
tg_request = tg_api.SendBytesPhotoRequest(
chat_id=chat_id,
photo=photo_content,
filename=photo_filename,
)
await tg_request.asend()
Пример отправки пользователю фото по URL:
import tg_api
async def main(token: str, chat_id: int, photo_filename: str, photo_url: str) -> None:
async with tg_api.AsyncTgClient.setup(token):
tg_request = tg_api.SendUrlPhotoRequest(
chat_id=chat_id,
photo=photo_url,
filename=photo_filename,
)
await tg_request.asend()
Пример отправки пользователю документа из файловой системы:
import aiofiles
import tg_api
async def main(token: str, chat_id: int, document_filename: str) -> None:
async with tg_api.AsyncTgClient.setup(token):
async with aiofiles.open(document_filename, 'rb') as f:
document_content = await f.read()
tg_request = tg_api.SendBytesDocumentRequest(
chat_id=chat_id,
document=document_content,
filename=document_filename,
)
await tg_request.asend()
Пример отправки пользователю документа по URL:
import tg_api
async def main(token: str, chat_id: int, document_filename: str, document_url: str) -> None:
async with tg_api.AsyncTgClient.setup(token):
tg_request = tg_api.SendUrlDocumentRequest(
chat_id=chat_id,
document=document_url,
filename=document_filename,
)
await tg_request.asend()
Пример поллинга:
import asyncio
from contextlib import suppress
from tg_api import AsyncTgClient, GetUpdatesRequest, Update
TG_BOT_TOKEN = "replace me with a real token"
async def handle_update(update: Update) -> None:
print(vars(update)) # your logic comes here
async def main() -> None:
async with AsyncTgClient.setup(TG_BOT_TOKEN):
tg_request = GetUpdatesRequest(timeout=30)
async for update in tg_request.alisten_to_updates():
await handle_update(update)
if __name__ == '__main__':
with suppress(KeyboardInterrupt):
asyncio.run(main())