Build a Python Script to Monitor API Errors and Auto‑Alert via Email

In a company that integrates many third‑party APIs, error stability varies and high traffic can cause frequent failures. To detect problematic interfaces quickly, a Python monitoring script is built to query the MySQL log database, count recent ERROR logs, and send email alerts when the count exceeds a set threshold.

1. Access the MySQL log database

def get_con():
    host = "127.0.0.1"
    port = 3306
    logsdb = "logsdb"
    user = "root"
    password = "never tell you"
    con = MySQLdb.connect(host=host, user=user, passwd=password, db=logsdb, port=port, charset="utf8")
    return con

2. Calculate the time window (last 2 minutes)

def calculate_time():
    now = time.mktime(datetime.now().timetuple()) - 60 * 2
    result = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(now))
    return result

3. Retrieve recent error logs

def get_data():
    select_time = calculate_time()
    logger.info("select time:" + select_time)
    sql = (
        "select file_name, message from logsdb.app_logs_record "
        "where log_time > '" + select_time + "' "
        "and level = 'ERROR' "
        "order by log_time desc"
    )
    conn = get_con()
    cursor = conn.cursor()
    cursor.execute(sql)
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    return results

4. Send email alerts

def send_email(content):
    sender = "[email protected]"
    receiver = ["[email protected]", "[email protected]"]
    host = 'smtp.163.com'
    port = 465
    msg = MIMEText(content)
    msg['From'] = sender
    msg['To'] = ",".join(receiver)
    msg['Subject'] = "system error warning"
    try:
        smtp = smtplib.SMTP_SSL(host, port)
        smtp.login(sender, '123456')
        smtp.sendmail(sender, receiver, msg.as_string())
        logger.info("send email success")
    except Exception as e:
        logger.error(e)

5. Periodic monitoring task

def task():
    while True:
        logger.info("monitor running")
        results = get_data()
        if results is not None and len(results) > 5:
            content = "recharge error:
"
            logger.info("a lot of error, so send mail")
            for r in results:
                content += r[1] + '
'
            send_email(content)
        time.sleep(2 * 60)

6. Configure logging

# coding=utf-8
import logging
logger = logging.getLogger('mylogger')
logger.setLevel(logging.DEBUG)
fh = logging.FileHandler('monitor.log')
fh.setLevel(logging.INFO)
ch = logging.StreamHandler()
ch.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
fh.setFormatter(formatter)
ch.setFormatter(formatter)
logger.addHandler(fh)
logger.addHandler(ch)

7. Run the monitor in a separate thread

# coding=utf-8
import threading
import MySQLdb
from datetime import datetime
import time
import smtplib
from email.mime.text import MIMEText
from log import logger

# (functions get_con, calculate_time, get_data, send_email, task defined above)

def run_monitor():
    monitor = threading.Thread(target=task)
    monitor.start()

if __name__ == "__main__":
    run_monitor()

8. Deploy with Supervisor

Install Supervisor on the server (e.g., CentOS 6) and add a program entry:

[program:app-monitor]
command=python /root/monitor/app_monitor.py
directory=/root/monitor
user=root

Start Supervisor with supervisord and manage the script using supervisorctl (e.g., status to check the process).

The monitoring script can be extended to watch specific interfaces, send SMS alerts, or integrate with other notification channels.