Telegram

Telegram is a cloud-based mobile and desktop messaging app with a focus on security and speed. For me, Telegram - is not just a messenger. It is the first app I open in the morning to read the news. And I am making a significant part of daily communication with Telegram. Plus, it is a developer-friendly service with an open API, so you can create your own clients and apps. More than, Telegram creators are encouraging it and have published a library to speed up that process.

Bots

Bots are third-party applications, hosted on your own server and run inside Telegram. Users can interact with bots by sending them messages, commands and inline requests. You implement the whole bots' logic and control their interaction with users using HTTPS requests to Telegram's Bot API. For example, if you have a website and monitor its availability. You can write a simple bot that will send you an alert if it's offline. There is a nice article about bots on the official website

Creating a bot

To create your own bot, you need to message a special bot - BotFather. In this article, we will create our own bot, that will implement a to-do list app. It should:

• keep user's notes in the database
• delete expired items (let's say that 1 hour is the expiry period)
• be able to fetch and send the whole list of saved notes

So, let's talk to BotFather:

Now, we have a bot token, it's everything we need. Let's dive into the code. Important code listings will be provided during the article. The whole repo link is at the bottom.

GINO ORM

First, let me introduce you to the other subject of this article - GINO ORM. Gino is a lightweight asynchronous ORM built on top of SQLAlchemy core for Python asyncio. GINO 1.1 supports PostgreSQL with asyncpg, and MySQL with aiomysql.

Create a new poetry project and add the required dependencies:

poetry new hashnode-todo
cd hashnode-todo
poetry add python-dotenv aiogram gino emoji

Then let's create models.py (and I would prefer to rename the project folder to "bot"):

from gino import Gino
from sqlalchemy import func

db = Gino()


class User(db.Model):
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    telegram_user_id = db.Column(db.String(length=200))
    created_at = db.Column(db.DateTime(timezone=True), default=func.now())
    first_name = db.Column(db.String(length=255))
    last_name = db.Column(db.String(length=255))
    username = db.Column(db.String(length=255))


class Note(db.Model):
    __tablename__ = 'notes'

    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'))
    created_at = db.Column(db.DateTime(timezone=True), default=func.now())
    text = db.Column(db.Text)

Also, we need to define a settings file:

import os
from dataclasses import dataclass
from dotenv import load_dotenv


load_dotenv()


@dataclass
class Settings:
    DATABASE_CONNECTION: str
    TELEGRAM_BOT_TOKEN: str


settings = Settings(
    DATABASE_CONNECTION=os.getenv('DATABASE_CONNECTION'),
    TELEGRAM_BOT_TOKEN=os.getenv('TELEGRAM_BOT_TOKEN'),
)

And simple .env file:

DATABASE_CONNECTION=postgresql://postgres@localhost/hashnode_todo
TELEGRAM_BOT_TOKEN=

Now we are ready to work with a database.

For this tutorial, we will initialize the database with gino.create_all() call. In real usage, the migration tool "Alembic" is recommended by GINO authors (and it was created by SQLAcheme author). Or you could assume, that other applications manage the database, so you only define the schema for the already existing database.

The database initialization file:

import asyncio

from bot.models import db
from config import settings


async def main():
    await db.set_bind(settings.DATABASE_CONNECTION)
    await db.gino.create_all()
    await db.pop_bind().close()


if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

Now you can run it with poetry run python bot/create_models.py

GINO queries

Let's think about the functions list, that we gonna need. And we know, that Telegram always has the user_id to identify the sender. So we need:

• add_note(user, text)
• get_notes(user)
• process_expired_notes()

And the simple service code with all required basic operations:

from datetime import datetime, timedelta
from typing import List

from bot.models import User, Note


async def add_note(user: 'User', text: str) -> 'Note':
    return await Note.create(
        user_id=user.id,
        text=text
    )


async def get_notes(user: 'User') -> List['Note']:
    return await Note.query.where(Note.user_id == user.id).gino.all()


async def process_expired_notes():
    await Note.delete.where(
        Note.created_at < datetime.now() - timedelta(days=1)
    ).gino.status()

Now we have all database management functionality, and we can use it in the bot

aiogram

aiogram is a pretty simple and fully asynchronous framework for Telegram Bot API written in Python 3.7 with asyncio and aiohttp.

If your bot needs to handle many requests, then asynchronous is your parachute.

Basic bot

Let's start with a simple bot, replying with "hello" on your /start command. The proposed structure is probably not ideal, but very scalable. I use it in some complex projects.

First, we need a base runner file main.py in the project root:

from aiogram import executor

from bot.models import db
from bot.main import dispatcher
from config import settings


async def on_startup(dispatcher):
    await db.set_bind(settings.DATABASE_CONNECTION)


async def on_shutdown(dispatcher):
    await db.pop_bind().close()


if __name__ == '__main__':
    executor.start_polling(
        dispatcher=dispatcher,
        skip_updates=True,
        on_startup=on_startup,
        on_shutdown=on_shutdown,
    )

And file for dispatcher and importing dialogs (bot/main.py):

from aiogram import Bot
from aiogram.dispatcher import Dispatcher
from config import settings


bot = Bot(token=settings.TELEGRAM_BOT_TOKEN)
dispatcher = Dispatcher(bot)
Dispatcher.set_current(dispatcher)


import bot.dialogs

Finally, the first dialog file for the "start" command:

from aiogram import types
from aiogram.dispatcher import Dispatcher


dp = Dispatcher.get_current()


@dp.message_handler(commands=['start'])
async def start_handler(message: types.Message):
    await message.answer('Hello from bot!')

Then just run it with poetry run python main.py

Now our simple bot can send the simple answer:

Middlewares

Middlewares are called on every message, so you can do throttling, logging or add variables to all future handlers. Let's add a middleware, that will get or create a user in the database.

First, adding a new service with one function - get_or_create_user:

from bot.models import User


async def get_or_create_user(
        user_id: str,
        first_name: str,
        last_name: str,
        username: str,
) -> ('User', bool):
    created = False
    user = await User.query.where(User.telegram_user_id == user_id).gino.first()
    if not user:
        user = await User.create(
            telegram_user_id=user_id,
            first_name=first_name,
            last_name=last_name,
            username=username
        )
        created = True

    return user, created

Second, the middleware itself:

from aiogram import types
from aiogram.dispatcher.middlewares import BaseMiddleware

from bot.services.users import get_or_create_user


class UserStorageMiddleware(BaseMiddleware):
    def __init__(self, database):
        self.database = database
        super().__init__()

    async def on_process_message(self, message: types.Message, data: dict):
        user = message.from_user
        db_user, _created = await get_or_create_user(
            user_id=str(user.id),
            first_name=user.first_name,
            last_name=user.last_name,
            username=user.username
        )
        data['user'] = db_user

Pay attention to conditionally-added variables to the data dictionary. If for some reasons, the 'user' is not set in the middleware, and you are declaring it in the handler, you will get an exception!

Third, add this middleware to the dispatcher:

dispatcher = Dispatcher(bot)
Dispatcher.set_current(dispatcher)

dispatcher.middleware.setup(UserStorageMiddleware(database=db))

And you can now use the user variable in handlers. Let's make our bot reply with the user's first name:

@dp.message_handler(commands=['start'])
async def start_handler(message: types.Message, user: 'User'):
    await message.answer(f'Hi, {user.first_name}!')

Keyboards and state

Keyboards are the last thing we need to implement for the planned note-taking bot. They allow you to show pre-defined actions and easily interact with the bot in one tap.

Let's add a file with a function to return the keyboard (bot/keyboards.py):

from aiogram import types
from aiogram.utils.emoji import emojize

from bot.texts import KEYBOARDS


def main_keyboard() -> types.ReplyKeyboardMarkup:
    markup = types.ReplyKeyboardMarkup(resize_keyboard=True)
    markup.row(
        types.KeyboardButton(emojize(KEYBOARDS['list_notes'])),
        types.KeyboardButton(emojize(KEYBOARDS['add_note']))
    )
    return markup

And a small file with texts:

RESPONSES = {
    'welcome': 'Hello, {user}, I am note taking bot for hashnode',
}

KEYBOARDS = {
    'list_notes': ':spiral_notepad: List notes',
    'add_note': ':heavy_plus_sign: Add note',
}

Plus, modify the start command handler:

@dp.message_handler(commands=['start'])
async def start_handler(message: types.Message, user: 'User'):
    await message.answer(
        RESPONSES['welcome'].format(user=user.first_name),
        reply_markup=main_keyboard(),
    )

And that simple steps allow you to make the bot more interactive:

Just 2 more steps to finish! Stay tuned :)

Pressing the button just sends its text. So we can use the Text filter in the message handler, to catch that event. Based on that, let's add another dialog file (bot/dialogs/list_notes.py), which will return all your notes

from aiogram import types
from aiogram.dispatcher import Dispatcher
from aiogram.dispatcher.filters import Text
from aiogram.utils.emoji import emojize

from bot.texts import KEYBOARDS
from bot.models import User
from bot.services import notes as notes_service
from bot.keyboards import main_keyboard


dp = Dispatcher.get_current()


@dp.message_handler(Text(equals=emojize(KEYBOARDS['list_notes'])))
async def handler_list_notes(message: types.Message, user: User):
    notes = await notes_service.get_notes(user)

    if not notes:
        answer_text = "You don't have any actual notes"
    else:
        notes_text = '\n\n'.join([
            f'{n.created_at.isoformat()}\n{n.text}' for n in notes
        ])
        answer_text = 'Your notes:\n\n' + notes_text

    await message.answer(
        answer_text,
        reply_markup=main_keyboard()
    )

For adding notes - we gonna use the state. If any state is activated, only related handlers can process the message. So we need to declare the State class and a handler, that will set it. Plus, we gonna replace the keyboard, showing only one "Cancel" button:

class AddNoteState(StatesGroup):
    enter_text = State()


@dp.message_handler(Text(equals=emojize(KEYBOARDS['add_note'])))
async def handler_add_note(message: types.Message):
    await AddNoteState.enter_text.set()
    await message.answer(
        RESPONSES['add_note']['prompt'],
        reply_markup=cancel_keyboard(),
    )

Then we just define the handler for the state, which will save the entered text as a new note:

@dp.message_handler(state=AddNoteState.enter_text)
async def handler_note_text(message: types.Message, state: FSMContext, user: 'User'):
    await notes_service.add_note(user, message.text)
    await state.finish()
    await message.answer(
        RESPONSES['add_note']['success'],
        reply_markup=main_keyboard(),
    )

A full resulting bot interaction:

A full source code is available on GitHub