Building real-time notification infrastructure is essential for modern web applications. At RecruitmentAlert, when an official Nigerian government portal opens a verified recruitment drive, thousands of subscribers receive instant alerts on their smartphones via our automated Telegram bot.
In this tutorial, you will learn step-by-step how to build a fully functional, production-ready Telegram notification bot using Python, Django REST Framework, and PostgreSQL.
Prerequisites & Architecture Overview
Before we start writing code, ensure you have:
- Python 3.10+ and Django installed in your virtual environment.
- A PostgreSQL database connected to your Django application.
- A public HTTPS URL (or Ngrok during local development) for receiving webhook requests from Telegram.
Our architecture consists of four main building blocks:
- Bot Creation: Obtaining an API Access Token from Telegram's BotFather.
- Database Model: Storing subscriber chat IDs and preference states in PostgreSQL.
- Webhook Handler View: A Django REST endpoint receiving incoming Telegram webhook JSON updates.
- Broadcasting Engine: A Python utility that iterates over active subscribers and broadcasts real-time alert messages via HTTP POST requests to Telegram's Bot API.
Step 1: Registering Your Bot with BotFather
Open your Telegram desktop or mobile app and search for @BotFather. Start a conversation and send the /newbot command:
# Telegram Chat Session with @BotFather
/newbot
# Response: Alright, a new bot. How are we going to call it? Please choose a name.
RecruitmentAlert Bot
# Response: Good. Now let's choose a username. It must end in `bot`.
govalerts_bot
# Response: Done! Congratulations on your new bot.
# Use this token to access the HTTP API:
# 7192847192:AAH9f2kLskP19823k_ExampleTokenHere
Save your token securely in your Django .env file:
TELEGRAM_BOT_TOKEN=7192847192:AAH9f2kLskP19823k_ExampleTokenHere
Step 2: Defining the Subscriber Model in PostgreSQL
In your Django app (e.g. apps/bot/models.py), create a database model to track users who interact with your bot.
# apps/bot/models.py
from django.db import models
from django.utils import timezone
class TelegramSubscriber(models.Model):
chat_id = models.BigIntegerField(unique=True, db_index=True, help_text="Telegram User/Chat ID")
username = models.CharField(max_length=150, blank=True, default='')
first_name = models.CharField(max_length=150, blank=True, default='')
is_active = models.BooleanField(default=True, db_index=True, help_text="Set false if user blocks the bot")
subscribed_at = models.DateTimeField(default=timezone.now)
class Meta:
db_table = 'telegram_subscribers'
ordering = ['-subscribed_at']
def __str__(self):
return f"{self.first_name} ({self.chat_id})"
Run Django migrations to create the database table in PostgreSQL:
python manage.py makemigrations bot
python manage.py migrate bot
Step 3: Writing the Webhook Handler View in Django
Telegram sends incoming messages as HTTP POST requests containing JSON updates. We will write a Django REST API view (APIView) decorated with @csrf_exempt to handle incoming /start and /help commands.
# apps/bot/views.py
import os
import logging
from django.views.decorators.csrf import csrf_exempt
from django.utils.decorators import method_decorator
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
import requests
from .models import TelegramSubscriber
logger = logging.getLogger(__name__)
BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
@method_decorator(csrf_exempt, name='dispatch')
class TelegramWebhookView(APIView):
def post(self, request):
data = request.data
if not data or "message" not in data:
return Response({"status": "ignored"}, status=status.HTTP_200_OK)
message = data["message"]
chat_id = message.get("chat", {}).get("id")
text = message.get("text", "").strip()
first_name = message.get("from", {}).get("first_name", "")
username = message.get("from", {}).get("username", "")
if not chat_id:
return Response({"status": "error"}, status=status.HTTP_400_BAD_REQUEST)
# Handle /start Command
if text.startswith("/start"):
subscriber, created = TelegramSubscriber.objects.get_or_create(
chat_id=chat_id,
defaults={
"first_name": first_name,
"username": username,
"is_active": True,
}
)
if not subscriber.is_active:
subscriber.is_active = True
subscriber.save()
welcome_msg = (
f"Hello {first_name}! Welcome to RecruitmentAlert.\n\n"
"You are now subscribed to receive instant alerts whenever "
"official Nigerian federal government recruitment portals open."
)
self.send_telegram_message(chat_id, welcome_msg)
return Response({"status": "ok"}, status=status.HTTP_200_OK)
def send_telegram_message(self, chat_id: int, text: str):
url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
payload = {
"chat_id": chat_id,
"text": text,
"parse_mode": "Markdown",
"disable_web_page_preview": False,
}
try:
requests.post(url, json=payload, timeout=5)
except Exception as e:
logger.error(f"Failed to send Telegram message to {chat_id}: {e}")
Register the view URL in your Django urls.py:
# apps/bot/urls.py
from django.urls import path
from .views import TelegramWebhookView
urlpatterns = [
path('webhook/', TelegramWebhookView.as_view(), name='telegram_webhook'),
]
Step 4: Registering the Webhook URL with Telegram API
To instruct Telegram to forward incoming updates to your Django endpoint, send a HTTP GET or POST request to Telegram's setWebhook endpoint:
# Register Webhook with Telegram
curl -X POST "https://api.telegram.org/bot7192847192:AAH9f2kLskP19823k_ExampleTokenHere/setWebhook" \
-H "Content-Type: application/json" \
-d '{"url": "https://www.recruitmentalert.com.ng/telegram/webhook/"}'
Response verification:
{
"ok": true,
"result": true,
"description": "Webhook was set"
}
Step 5: Broadcasting Real-Time Notifications
When a new verified recruitment opening is created in Django, call this broadcasting service function to deliver alerts to all active subscribers.
# apps/bot/services.py
import os
import requests
import logging
from .models import TelegramSubscriber
logger = logging.getLogger(__name__)
BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
def broadcast_recruitment_alert(job_title: str, agency_name: str, portal_url: str):
subscribers = TelegramSubscriber.objects.filter(is_active=True)
success_count = 0
failure_count = 0
message_text = (
f"🚨 **VERIFIED RECRUITMENT ALERT**\n\n"
f"**Agency:** {agency_name}\n"
f"**Position:** {job_title}\n\n"
f"**Official Portal:** [Apply Here]({portal_url})\n\n"
f"⚡ *Government recruitment is 100% free. Never pay for job forms.*"
)
api_url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
for sub in subscribers:
payload = {
"chat_id": sub.chat_id,
"text": message_text,
"parse_mode": "Markdown",
"disable_web_page_preview": True,
}
try:
res = requests.post(api_url, json=payload, timeout=5)
if res.status_code == 200:
success_count += 1
elif res.status_code == 403:
sub.is_active = False
sub.save()
failure_count += 1
else:
failure_count += 1
except Exception as err:
logger.error(f"Error broadcasting to {sub.chat_id}: {err}")
failure_count += 1
return {"success": success_count, "failed": failure_count}
Conclusion
You now have a production-ready, asynchronous notification system integrated with Python, Django, PostgreSQL, and Telegram. This architecture ensures high deliverability, automated subscriber state management, and real-time alerts for thousands of users.