自动化数据库运维双11优惠活动

import psycopg2
import smtplib
from email.mime.text import MIMEText

def check_db_connections(threshold):
    try:
        conn = psycopg2.connect(database="yourdb", user="youruser", password="yourpass", host="yourhost")
        cur = conn.cursor()
        cur.execute("SELECT count(*) FROM pg_stat_activity;")
        active_connections = cur.fetchone()[0]
        cur.close()
        conn.close()
        
        if active_connections > threshold:
            send_alert_email(active_connections)
    except Exception as e:
        print(f"Error checking database connections: {e}")

def send_alert_email(connections):
    msg = MIMEText(f"Database connection limit exceeded: {connections}")
    msg['Subject'] = 'Database Connection Alert'
    msg['From'] = 'alert@example.com'
    msg['To'] = 'admin@example.com'
    
    s = smtplib.SMTP('localhost')
    s.send_message(msg)
    s.quit()

# Example usage
check_db_connections(100)