The Agentic Database Backup Pipeline
Database backups are essential for disaster recovery: if your database is corrupted, deleted, or compromised, you need a backup to restore from. But manual backups are tedious and error-prone: you have to remember to run them, verify they succeeded, and test the restore process. Many teams skip backups entirely, which means a database failure can destroy their business. What you need is an agentic database backup pipeline that backs up your database automatically, verifies the backups, and tests the restore process. With the Deployxa MCP server, this is straightforward to build. Here is how.
The direct answer is that a database backup pipeline is a script that runs on a schedule (e.g., daily), creates a backup of your database (via pg_dump for Postgres), uploads the backup to external storage (e.g., S3, Cloudflare R2), verifies the backup (by checking its size and integrity), and periodically tests the restore process (by restoring the backup to a test database and verifying the data). The pipeline uses the Deployxa MCP server to coordinate the backup process and to notify you of issues. For more on database management, see our article on database connection pooling across blue/green deployments.
Why Manual Backups Are Dangerous
Three problems make manual backups dangerous. First, they are forgotten. Backups are a low-priority task, which means they get postponed indefinitely, and when you need a backup, you do not have one. Second, they are untested. A backup that has never been tested might not restore correctly, which means you discover at the worst possible moment that your backup is corrupt or incomplete. Third, they are not automated. Manual backups require a human to run them, which means they are not run consistently, and they are not run at 3 AM when nobody is awake. An agentic backup pipeline solves all three problems: it runs automatically (on a schedule), it tests restores (periodically), and it runs at any time (including 3 AM). For more on agentic patterns, see our article on building an AI agent that monitors your app 24/7.
What the Backup Pipeline Does
The agentic database backup pipeline performs the following tasks:
- Create a backup. On a schedule (e.g., daily), the pipeline creates a backup of the database using pg_dump (for Postgres). The backup is saved as a compressed file.
- Upload to external storage. The pipeline uploads the backup to external storage (e.g., S3, Cloudflare R2), which ensures the backup survives even if the database server is destroyed.
- Verify the backup. The pipeline verifies the backup by checking its size (a backup that is significantly smaller than usual might be incomplete) and by listing its contents (via pg_restore --list).
- Test the restore. Periodically (e.g., weekly), the pipeline tests the restore process by restoring the backup to a test database and verifying the data (e.g., checking the row counts match).
- Clean up old backups. The pipeline deletes backups older than a retention period (e.g., 30 days), to avoid unlimited storage growth.
- Notify on issues. The pipeline sends a Slack notification if a backup fails, if a verification fails, or if a restore test fails.
Step-by-Step: Building the Backup Pipeline
Here is how to build the database backup pipeline in Python.
Step 1: Install dependencies
pip install boto3 requests schedule
npm install -g @deployxa/mcp-server
deployxa-mcp loginStep 2: Create the backup script
# backup.py
import boto3
import subprocess
import os
import schedule
import time
import json
from datetime import datetime, timedelta
import requests
# Configuration
DATABASE_URL = os.environ.get("DATABASE_URL")
S3_BUCKET = os.environ.get("S3_BUCKET")
S3_REGION = os.environ.get("S3_REGION")
SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL")
DEPLOYXA_MCP_URL = "http://localhost:3000/mcp"
RETENTION_DAYS = 30
BACKUP_INTERVAL_HOURS = 24
RESTORE_TEST_INTERVAL_DAYS = 7
# S3 client
s3 = boto3.client("s3", region_name=S3_REGION)
def send_slack_notification(message, is_alert=False):
emoji = "💾" if not is_alert else "🚨"
requests.post(SLACK_WEBHOOK_URL, json={"text": f"{emoji} {message}"})
def create_backup():
timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
backup_filename = f"backup_{timestamp}.sql.gz"
print(f"Creating backup: {backup_filename}")
# Create the backup using pg_dump
try:
result = subprocess.run(
["pg_dump", DATABASE_URL, "-c", "-O", "|", "gzip", ">", backup_filename],
shell=True,
capture_output=True,
text=True,
)
if result.returncode != 0:
send_slack_notification(
f"Backup failed: {result.stderr}",
is_alert=True
)
return None
# Check the backup size
backup_size = os.path.getsize(backup_filename)
if backup_size < 1000: # Less than 1KB is suspicious
send_slack_notification(
f"Backup is suspiciously small ({backup_size} bytes). Check the database.",
is_alert=True
)
return None
print(f" Backup created: {backup_size} bytes")
return backup_filename
except Exception as e:
send_slack_notification(f"Backup failed: {e}", is_alert=True)
return None
def upload_to_s3(backup_filename):
print(f"Uploading {backup_filename} to S3...")
try:
s3.upload_file(backup_filename, S3_BUCKET, backup_filename)
print(f" Uploaded to s3://{S3_BUCKET}/{backup_filename}")
return True
except Exception as e:
send_slack_notification(f"Upload failed: {e}", is_alert=True)
return False
def verify_backup(backup_filename):
print(f"Verifying {backup_filename}...")
# List the backup contents to verify it's valid
try:
result = subprocess.run(
["gzcat", backup_filename, "|", "pg_restore", "--list"],
shell=True,
capture_output=True,
text=True,
)
if result.returncode != 0:
send_slack_notification(
f"Backup verification failed: {result.stderr}",
is_alert=True
)
return False
# Check if the backup contains tables
if "; Entry for" not in result.stdout:
send_slack_notification(
"Backup verification failed: no entries found",
is_alert=True
)
return False
print(f" Backup verified")
return True
except Exception as e:
send_slack_notification(f"Verification failed: {e}", is_alert=True)
return False
def test_restore(backup_filename):
print(f"Testing restore from {backup_filename}...")
test_db_url = os.environ.get("TEST_DATABASE_URL")
if not test_db_url:
print(" No TEST_DATABASE_URL set, skipping restore test")
return
try:
# Download the backup from S3
s3.download_file(S3_BUCKET, backup_filename, f"/tmp/{backup_filename}")
# Restore to the test database
result = subprocess.run(
["gzcat", f"/tmp/{backup_filename}", "|", "psql", test_db_url],
shell=True,
capture_output=True,
text=True,
)
if result.returncode != 0:
send_slack_notification(
f"Restore test failed: {result.stderr}",
is_alert=True
)
return
# Verify the data (check row counts)
# In a real implementation, you would compare row counts
# between the production and test databases
print(f" Restore test successful")
send_slack_notification("Restore test successful")
except Exception as e:
send_slack_notification(f"Restore test failed: {e}", is_alert=True)
def cleanup_old_backups():
print("Cleaning up old backups...")
cutoff = datetime.utcnow() - timedelta(days=RETENTION_DAYS)
# List all objects in the S3 bucket
response = s3.list_objects_v2(Bucket=S3_BUCKET)
for obj in response.get("Contents", []):
key = obj["Key"]
last_modified = obj["LastModified"].replace(tzinfo=None)
if last_modified < cutoff:
print(f" Deleting {key} (last modified: {last_modified})")
s3.delete_object(Bucket=S3_BUCKET, Key=key)
def run_backup():
print(f"[{datetime.utcnow().isoformat()}] Running database backup...")
# Step 1: Create backup
backup_filename = create_backup()
if not backup_filename:
return
# Step 2: Upload to S3
if not upload_to_s3(backup_filename):
return
# Step 3: Verify backup
if not verify_backup(backup_filename):
return
# Step 4: Clean up local file
os.remove(backup_filename)
# Step 5: Notify success
send_slack_notification(f"Database backup successful: {backup_filename}")
# Step 6: Clean up old backups
cleanup_old_backups()
def run_restore_test():
print(f"[{datetime.utcnow().isoformat()}] Running restore test...")
# Get the latest backup
response = s3.list_objects_v2(Bucket=S3_BUCKET)
if not response.get("Contents"):
print(" No backups found")
return
latest_backup = max(response["Contents"], key=lambda x: x["LastModified"])
test_restore(latest_backup["Key"])
# Schedule backup (daily) and restore test (weekly)
schedule.every(BACKUP_INTERVAL_HOURS).hours.do(run_backup)
schedule.every(RESTORE_TEST_INTERVAL_DAYS).days.do(run_restore_test)
# Run immediately, then on schedule
run_backup()
while True:
schedule.run_pending()
time.sleep(1)Step 3: Set environment variables
In the Deployxa dashboard, set:
- DATABASE_URL: your production database URL
- TEST_DATABASE_URL: a test database URL (for restore testing)
- S3_BUCKET: your S3 bucket name
- S3_REGION: your S3 region
- SLACK_WEBHOOK_URL: your Slack webhook URL
Step 4: Run the pipeline
python backup.pyThe pipeline runs a backup immediately, then every 24 hours. It also runs a restore test every 7 days.
Step 5: Deploy the pipeline
The pipeline should run on a server (not your laptop), so it runs 24/7. You can deploy it as a Deployxa app.
Common Pitfalls and Troubleshooting
The first pitfall is not testing restores. A backup that has never been restored might not work, which means you discover at the worst possible moment that your backup is corrupt. The fix is to test restores regularly (e.g., weekly). The second pitfall is not verifying backups. A backup that is incomplete or corrupt is useless, which means you need to verify each backup (e.g., check its size, list its contents). The fix is to verify each backup after creation. The third pitfall is not cleaning up old backups. Backups accumulate over time, which can fill up your storage. The fix is to set a retention period (e.g., 30 days) and to delete old backups. The fourth pitfall is storing backups in the same location as the database. If the database server is destroyed, the backups are destroyed too. The fix is to store backups in external storage (e.g., S3, Cloudflare R2), which is in a different location. The fifth pitfall is not securing backups. Backups contain sensitive data (e.g., user passwords, API keys), which means they need to be encrypted. The fix is to encrypt backups before uploading them (e.g., using gpg or S3's server-side encryption).
Conclusion: Back Up Automatically, Test Restores Regularly
Manual backups are forgotten, untested, and not automated. An agentic database backup pipeline that backs up automatically, verifies backups, and tests restores is the sustainable solution. With the Deployxa MCP server and Python, you can build a backup pipeline that protects your data and gives you peace of mind. Stop running manual backups and start automating.
Ready to build your backup pipeline? Install the Deployxa MCP server with npm i -g @deployxa/mcp-server, run deployxa-mcp login, and start building. For more on agentic workflows, see our articles on building an AI agent that manages your secrets and the agentic blue/green deployment pipeline. Learn about building an AI agent that optimizes your database and the agentic log analysis pipeline in our companion articles. Explore our free developer tools to speed up your workflow.