Telegram task manager bot with MongoDB
Introduction
A year ago, while I was learning to code in python, I had made a simple console to-do-list app. It was a project from the EduTools plugin for the PyCharm. The outcomes like the basis of SQLAlchemy and SQLite were quite impactful. Lastly, I came up with an idea to make a to-do-list app again but in a very different way, meanwhile describing the process in a blog post. As a Telegram power user, I am interested in building bots for this communicator. So instead of writing a standard flask web app, I would like to use a telegram bot to manage the tasks. Recently, I have been also reading a lot about NoSql databases, so I want to use MongoDB as the bot’s database. It is not the best choice for such a small app (I would use just SQLite). However, I have never used MongoDB, so learning value is pretty worth it for me. I am familiar with Heroku (and Heroku’s fan) so it is a simple choice to deploy the bot.
To-do bot’s features
First of all, my needs are quite little, so I do not want such a complicated bot. Here is a brief and illustrative list of the bot’s features:
- adding new tasks to the database (fields: date, task, category, deadline, status),
- ending tasks (one or multiple at once),
- searching the database:
- show all tasks,
- show deadlines (tasks with deadline date in the next 24 hours),
- show tasks with specified category,
- show tasks with the specified status.
Non-blog steps
I am not going to show creating a new telegram bot with BotFather, deploying a database with MongoDB Atlas, creating a new app in Heroku, or creating a new virtual environment with conda which I use as a package manager (anyway I am going to show this in Dockerfile). Both things are well described in many sources. To be honest, the post is not a tutorial, so I see no point in showing those steps.
Building the bot
After a short talk with BotFather, a few clicks in MongoDB Atlas and Heroku almost everything is ready to start scripting the bot. I also have to create some files in my project’s directory.
$ touch main.py handlers.py database.py config.py helpers.py entrypoint.sh .gitignore Dockerfile
Every file except .gitignore will be mentioned in the rest of the post, no worries. After creating a virtual environment I also need to export the dependency yaml file. It will be useful in the future.
$ conda env export > environment.yml
Managing environment variables
The environment variables I have to configure are the telegram token and MongoDB’s connection string. I want to store the name of the Heroku app in the environment variable also (it is not necessary but I think it is a good practice). In Heroku config variables can be managed via Heroku CLI or Heroku Dashboard. I usually do this via Heroku CLI like this:
$ heroku config: set TOKEN=<telegram token>
$ heroku config: set MONGO_ACCESS=<mongodb connection string>
$ heroku config: set HEROKU_APP_NAME=<heroku app name>
I also add the following code to the config.py to get environment variables using the built-in os module.
import os
TOKEN = os.getenv('TOKEN')
MONGO_ACCESS = os.getenv('MONGO_ACCESS')
PORT = int(os.environ.get("PORT", 8443))
HEROKU_APP_NAME = os.environ.get("HEROKU_APP_NAME")
Preparing the database
After creating the database and collections in MongoDB Atlas I can add some basic configuration code to the database.py file.
import pymongo
from config import MONGO_ACCESS
# configure mongodb client
client = pymongo.MongoClient(MONGO_ACCESS)
db = client.tasksdb
tasks = db.todos
# function that gives the next sequential ID for a collection by name
def get_sequence(name) -> dict:
seq = db.sequences
doc = seq.find_one_and_update(
{'_id': name}, {'$inc': {'value': 1}}, return_document=True)
return doc['value']
The get_sequence function is essential for the usage of the bot. The user has an option to end the tasks by providing their unique ids. The default _id object generated by MongoDB is not easy to read or rewrite. I would like to have simple numeric ids. MongoDB does not have a built-in auto-incrementing function, so there is a need to implement it on my own. The most popular workaround is creating a new collection that holds a last used id for the collection and making a function that can give the next sequential id by the collection’s name.
Managing bot’s interactions
Interaction with the user is a core of the chatbot. To manage the interactions the handler methods should be created. I will do it in handlers.py file.
import re
from datetime import datetime, timedelta
from telegram import ReplyKeyboardRemove, InlineKeyboardButton, \
InlineKeyboardMarkup, ParseMode, Update
from telegram.ext import ConversationHandler, CallbackContext
from database import tasks, get_sequence
from helpers import search_flow_helper
# define options
CHOOSE, STATUS, ADD_NEW, END_TASK, SEARCH = range(5)
The variables after importing dependencies are representing the states the bot can be as users interact with it. After defining them I can start adding the handler methods to the script. The first one is pretty obvious, the bot must be started somehow. Standard /start command initiates the conversation and sends a welcome message.
# starting state of the bot
def start(update, context) -> int:
bot = context.bot
chat_id = update.message.chat.id
bot.send_message(chat_id=chat_id,
text="Hi, I'm Task Manager Bot. I'm here to help you stop being lazy.\n"
"Please type /choose to show a menu")
return CHOOSE
In the message, the bot asks the user to type /choose to show a menu and the function returns CHOOSE state. It is tied to the handler method - choose, which displays three buttons: ADD NEW, END TASK, and SEARCH. This state is the heart of the bot. After doing every operation user has an option to back to this menu or exit the bot.
# CHOOSE state
def choose(update, context) -> int:
bot = context.bot
chat_id = update.message.chat.id
data = update.message.text
# if input is not correct end a conversation
if data != '/choose':
invalid_entry_handler(bot, chat_id)
else:
# reply keyboard with buttons - main menu
reply_keyboard = [
[
InlineKeyboardButton(text='ADD NEW', callback_data='ADD NEW')
],
[
InlineKeyboardButton(text='END TASK', callback_data='END TASK')
],
[
InlineKeyboardButton(text='SEARCH', callback_data='SEARCH')
]
]
markup = InlineKeyboardMarkup(reply_keyboard, one_time_keyboard=True)
bot.send_message(chat_id=chat_id, text='Please tell me what do you want to do',
reply_markup=markup)
return STATUS
The simple function invalid_entry_handler sends a message after providing invalid command by user. Every function with repeatable code is stored in helpers.py file.
def invalid_entry_handler(bot, chat_id):
return bot.send_message(
chat_id=chat_id, text='Invalid entry. Please type /start to restart a bot, /choose to show menu or /exit to terminate a bot')
Returned state - STATUS is responsible for managing the user’s choice. Depends on the button clicked it sends a brief instruction for the chosen feature and returns the proper state.
# STATUS state
def status(update, context) -> int:
bot = context.bot
chat_id = update.callback_query.message.chat.id
if update.callback_query.data == 'ADD NEW':
bot.send_message(chat_id=chat_id,
text='Great, please provide: task, category and deadline'
' (separated by semicolons, eg. Do the shopping; Household; 2021-01-01 15:15)')
return ADD_NEW
elif update.callback_query.data == 'END TASK':
bot.send_message(chat_id=chat_id,
text='Please provide end with list of tasks ids you would like to end (separated by spaces, eg. end 4 15 18 12)')
return END_TASK
elif update.callback_query.data == 'SEARCH':
bot.send_message(chat_id=chat_id,
text='Great, please take a look at brief instruction and type:\n'
'`show all` to show all tasks\n'
'`show deadlines` to show tasks with deadline in the next 24 hours\n'
'`show category {your category}` to show all tasks with specified category\n'
'`show status {done/not done yet}` to show all tasks with specified status',
parse_mode=ParseMode.MARKDOWN_V2)
return SEARCH
The next functions are responsible for the core bot’s features. The first is the add_new function, which is connected with the ADD_NEW state. This function parses the user’s input into the document and inserts it into the database. The user must provide the task’s name, category, and deadline’s date in the correct format. Id of the task is generated with the help of the get_sequence function. The date field is just the date of the message. The status field is set to ‘Not done yet’ by default. Username and user_id fields are skipped because I am not going to share the bot with other users. Otherwise, I would obviously implement it.
# ADD_NEW state
def add_new(update, context) -> int:
chat_id = update.message.chat_id
bot = context.bot
data = update.message.text.split('; ')
# check if input is correct and making a doc with proper datatypes is possible - tbh only the date can be a problem
if len(data) == 3:
try:
(current_time, task, category, deadline) = \
(update.message.date, data[0], data[1],
datetime.strptime(data[2], '%Y-%m-%d %H:%M'))
doc = {
'_id': get_sequence('todos'), # generate next sequential id
'date': current_time,
'task': task,
'category': category,
'deadline': deadline,
'status': 'Not done yet',
}
tasks.insert_one(doc) # insert to database
bot.send_message(chat_id=chat_id,
text='Success. Type /choose to show menu or /exit to disable me'
)
except:
invalid_entry_handler(bot, chat_id)
else:
invalid_entry_handler(bot, chat_id)
return CHOOSE
The search function allows querying the database with four commands:
- show all - find all records from the collection
- show deadlines - find all records with the deadline’s date within the next 24 hours
- show category {your category} - find all records with specified category’s name
- show status {done/not done yet} - find all record with specified status
Being obligated to remember if desired category or status is capitalized in the database would be a huge problem for me, so the function matches the exact string and ignores the case using a simple regex.
# SEARCH state
def search(update, context) -> int:
chat_id = update.message.chat_id
bot = context.bot
data = update.message.text
if data == 'show all': # find all results
cur = tasks.find({})
search_flow_helper(bot, cur, chat_id)
elif data == 'show deadlines': # find results with deadline in next 24 hours and not done status
today = datetime.today()
tommorow = today + timedelta(days=1)
cur = tasks.find({'$and': [{'deadline': {'$gte': today, "$lt": tommorow}}, {
'status': {'$ne': 'Done'}}]})
search_flow_helper(bot, cur, chat_id)
# find results with specified category and ignorcase
elif 'show category' in data and len(data.split()) == 3:
category = data.split()[2]
cur = tasks.find({'category': re.compile(
'^' + category + '$', re.IGNORECASE)}) # matching exact string
search_flow_helper(bot, cur, chat_id)
# find results with specified status and ignorcase
elif data in ['show status done', 'show status not done yet']:
status = data.partition('status ')[2]
cur = tasks.find({'status': re.compile(
'^' + status + '$', re.IGNORECASE)})
search_flow_helper(bot, cur, chat_id)
else:
invalid_entry_handler(bot, chat_id)
return CHOOSE
The search_flow_helper is another function that contains repeatable code. If there are no results for the query it sends an adequate message, otherwise, it transforms found documents into pretty strings ready to be displayed in the telegram’s chat window (for convenience it is in doc_to_string function).
# function that transforms found documents to string ready to be displayed by bot
def doc_to_string(cur) -> str:
doc = list(cur)
outside_list = [
['Id, Date of assignment, Task, Category, Deadline', 'Status']]
for x in doc:
inside_list = []
for k, v in x.items():
inside_list.append(str(v))
outside_list.append(inside_list)
return '\n'.join(', '.join(x) for x in outside_list) + '\nSuccess. Type /choose to show menu or /exit to disable me'
# function that help with repeatable checking if cursor is not empty and send right message to the user
def search_flow_helper(bot, cur, chat_id):
if len(list(cur.clone())) > 0:
return bot.send_message(chat_id=chat_id, text=doc_to_string(cur))
else:
return bot.send_message(
chat_id=chat_id, text='No results for your query\n'
'Type /choose to show menu or /exit to disable me')
The last feature I have to implement is ending tasks. Having an option to end multiple tasks is important for me. It has bad sides, for example, the user could mix the ids of ended and not ended tasks in one command. That’s why I want the bot to check if ids are correct and send success or error messages.
# END_TASK state
def end_task(update, context) -> int:
chat_id = update.message.chat_id
bot = context.bot
data = update.message.text.split()
# check if user input is correct - 'end' followed by digits
if data[0] == 'end' and all(x.isdigit() for x in data[1:]):
# get all task ids with status not done yet
cur = list(tasks.find({'status': 'Not done yet'}).distinct('_id'))
user_input_tasks = [int(x)
for x in data[1:]] # get ids provided by user
# get ids of tasks that cant be ended
bad_ids = [x for x in user_input_tasks if x not in cur]
# get id of tasks that can be ended
good_ids = [x for x in user_input_tasks if x in cur]
# if user provided bad ids show proper communicate
if bad_ids:
bot.send_message(
chat_id=chat_id, text=f"Tasks: {', '.join(str(x) for x in bad_ids)}"
" are already done or do not exists")
# for good ids update status and show communicate
if good_ids:
filter = {'_id': {'$in': good_ids}}
values = {'$set': {'status': 'Done'}}
tasks.update_many(filter, values)
bot.send_message(
chat_id=chat_id, text=f"Tasks: {', '.join(str(x) for x in good_ids)} ended successfully")
bot.send_message(
chat_id=chat_id, text='Type /choose to show menu or /exit to disable me')
else:
invalid_entry_handler(bot, chat_id)
return CHOOSE
The last thing I need to implement in handlers.py file is exit function. It just allows user to cancel conversation and will be tied to /exit command.
def exit(update: Update, context: CallbackContext) -> int:
update.message.reply_text(
'Bye! Type /start to wake me up whenever you want :).',
reply_markup=ReplyKeyboardRemove()
)
return ConversationHandler.END
ConversationHandler
Telegram’s ConversationHandler takes care of the state user is currently at. Every handler method from handlers.py file must be mapped to each state in the ConversationHandler. There is also a need to set a webhook with my heroku’s app configuration in the main.py file. I use webhook instead of polling because of efficiency. Polling sends a request at a specified frequency to check if there are some changes in the data. Webhooks are getting notified when the data is changed. It is more efficient and the bot has data in real-time. I would also like to use /exit command regardless of the state I am currently at, so it is necessary to exclude /exit from MessageHandlers.
from telegram.ext import CommandHandler, ConversationHandler, \
MessageHandler, Filters, Updater, CallbackQueryHandler
import handlers
from config import TOKEN, HEROKU_APP_NAME, PORT
def main():
updater = Updater(token=TOKEN, use_context=True)
dispatcher = updater.dispatcher
port = int(PORT)
# set heroku webhook
updater.start_webhook(listen="0.0.0.0",
port=port,
url_path=TOKEN, webhook_url="https://{}.herokuapp.com/{}".format(HEROKU_APP_NAME, TOKEN))
# filter to exclude /exit command in message handlers
exit_filter = (~ Filters.text(['/exit']))
conv_handler = ConversationHandler(
entry_points=[CommandHandler('start', handlers.start)],
states={
handlers.CHOOSE: [MessageHandler(Filters.text & exit_filter, handlers.choose)],
handlers.STATUS: [CallbackQueryHandler(handlers.status)],
handlers.ADD_NEW: [MessageHandler(Filters.text & exit_filter, handlers.add_new)],
handlers.SEARCH: [MessageHandler(Filters.text & exit_filter, handlers.search)],
handlers.END_TASK: [MessageHandler(
Filters.text & exit_filter, handlers.end_task)]
},
fallbacks=[CommandHandler('exit', handlers.exit)],
allow_reentry=True
)
dispatcher.add_handler(conv_handler)
if __name__ == '__main__':
main()
Docker and Heroku
Now it is time to make the bot alive. I will use Heroku to deploy the bot. Heroku is a platform as a service (PaaS) that enables developers to build, run, and operate applications entirely in the cloud. There are several ways to deploy the application. I prefer using Heroku Container Registry. It requires Docker to be installed. This is my Dockerfile, which will be used to build an image and push it to Heroku.
FROM continuumio/miniconda3
WORKDIR /task_manager_bot
# make conda env
COPY environment.yml .
RUN conda env create -f environment.yml
# Code to run when container starts
COPY config.py database.py handlers.py helpers.py main.py entrypoint.sh ./
ENTRYPOINT ["sh","./entrypoint.sh"]
As you can see there is conda environment created but not activated. It happens in the entrypoint.sh
#!/bin/bash --login
eval "$(command conda 'shell.bash' 'hook' 2 \
> /dev/null)"
conda activate task_manager_bot
exec python main.py
The first command is required to activate conda through a shell script, more to read here. Now it is possible to build an image and push it to Heroku via Heroku CLI by a simple command:
heroku container:push --app <HEROKU_APP_NAME> web
After a few minutes, the bot should be alive. The logs to confirm everything is fine can be printed to console by a command:
heroku logs --tail --app <HEROKU_APP_NAME>
Conclusion
Developing a telegram bot was fun for me, even if the bot was not so complicated. Maybe I could create more data-related bots in the future. You can also take a look at some screens that show the interactions with the bot.
