Building a Telegram Bot for Order and Inventory Management

This architecture uses Python, SQLite, and the Telegram Bot API to implement real-time inventory tracking and order handling for a small retail setup.

1. Database Schema Design

The state must be stored relationally. Atomic SQL transactions ensure inventory levels remain accurate during simultaneous purchasing attempts.

Create a database initialization script (setup.py):

import sqlite3

def init_db():
    conn = sqlite3.connect('shop.db')
    cursor = conn.cursor()
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS inventory (
            product_id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT UNIQUE NOT NULL,
            stock INTEGER NOT NULL CHECK(stock >= 0),
            price REAL NOT NULL
        )
    ''')
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS orders (
            order_id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER NOT NULL,
            product_id INTEGER NOT NULL,
            quantity INTEGER NOT NULL,
            status TEXT DEFAULT 'PENDING',
            FOREIGN KEY (product_id) REFERENCES inventory(product_id)
        )
    ''')
    conn.commit()
    conn.close()

if __name__ == '__main__':
    init_db()

2. Telegram Bot Configuration

• Open Telegram and start a chat with @BotFather.
• Send /newbot, follow the prompts, and record your HTTP API token.
• Install the Python Telegram framework: pip install pyTelegramBotAPI.

3. Core Bot Implementation

This script processes catalog requests and handles purchasing logic within database transactions to eliminate race conditions.

import sqlite3
import telebot

TOKEN = 'YOUR_BOT_TOKEN_HERE'
ADMIN_IDS = [123456789]  # Replace with actual Telegram User IDs

bot = telebot.TeleBot(TOKEN)

def get_db():
    return sqlite3.connect('shop.db')

@bot.message_handler(commands=['inventory'])
def list_inventory(message):
    conn = get_db()
    cursor = conn.cursor()
    cursor.execute("SELECT product_id, name, stock, price FROM inventory WHERE stock > 0")
    items = cursor.fetchall()
    conn.close()

    if not items:
        bot.reply_to(message, "Out of stock.")
        return

    text = "Available Products:\n"
    for item in items:
        text += f"ID: {item[0]} | {item[1]} - {item[2]} in stock | ${item[3]:.2f}\n"
    bot.reply_to(message, text)

@bot.message_handler(commands=['order'])
def place_order(message):
    # Syntax: /order  
    args = message.text.split()
    if len(args) != 3:
        bot.reply_to(message, "Usage: /order <product_id> <quantity>")
        return

    product_id, qty = int(args[1]), int(args[2])
    conn = get_db()
    cursor = conn.cursor()

    try:
        cursor.execute("BEGIN TRANSACTION")
        cursor.execute("SELECT stock FROM inventory WHERE product_id = ?", (product_id,))
        row = cursor.fetchone()

        if not row or row[0] < qty:
            bot.reply_to(message, "Order failed: Insufficient stock or invalid ID.")
            conn.rollback()
            return

        cursor.execute("UPDATE inventory SET stock = stock - ? WHERE product_id = ?", (qty, product_id))
        cursor.execute("INSERT INTO orders (user_id, product_id, quantity) VALUES (?, ?, ?)",
                       (message.from_user.id, product_id, qty))
        
        conn.commit()
        bot.reply_to(message, f"Success! Order placed for {qty} unit(s).")
    except Exception as e:
        conn.rollback()
        bot.reply_to(message, "System error during transaction.")
    finally:
        conn.close()

@bot.message_handler(commands=['addstock'])
def add_stock(message):
    if message.from_user.id not in ADMIN_IDS:
        bot.reply_to(message, "Unauthorized.")
        return

    # Syntax: /addstock   
    try:
        _, name, qty, price = message.text.split()
        conn = get_db()
        cursor = conn.cursor()
        cursor.execute('''
            INSERT INTO inventory (name, stock, price) VALUES (?, ?, ?)
            ON CONFLICT(name) DO UPDATE SET stock = stock + ?
        ''', (name, int(qty), float(price), int(qty)))
        conn.commit()
        conn.close()
        bot.reply_to(message, f"Updated stock for '{name}'.")
    except Exception:
        bot.reply_to(message, "Usage: /addstock <name> <quantity> <price>")

bot.infinity_polling()

4. Operational Deployment Requirements

• Process Management: Use systemd or Docker to run the script as a daemon, ensuring automatic restart on crashes.
• Backups: Schedule periodic backups of shop.db using a cron job.
• Scaling Limit: SQLite works for low to moderate traffic. Switch to PostgreSQL if handling simultaneous, high-frequency writes.

Need this done fast? order a Telegram bot on FreelanceHunt.