如何基于Python-Flask实现PostgreSQL数据库每日自动备份至云盘
Hey there! I’ve built similar backup workflows for Flask-PostgreSQL apps before, so let me walk you through a solid, actionable setup with code snippets you can adapt right away. Here’s how to automate daily backups to Google Drive or Dropbox:
First, we’ll generate a PostgreSQL backup using pg_dump, then upload the file to your chosen cloud storage via their Python SDKs, and finally set up a daily schedule to run the whole process automatically.
We’ll use Python’s subprocess to call pg_dump (make sure it’s installed on your server). This function generates a timestamped backup file to avoid overwriting old ones:
import subprocess import os from datetime import datetime def create_postgres_backup(): # Pull DB config from environment variables (never hardcode!) db_user = os.getenv("DB_USER") db_name = os.getenv("DB_NAME") db_password = os.getenv("DB_PASSWORD") # Generate unique backup filename with timestamp timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_file = f"postgres_backup_{timestamp}.sql" # Set PGPASSWORD env var to avoid password prompts env = os.environ.copy() env["PGPASSWORD"] = db_password try: # Execute pg_dump command subprocess.run( [ "pg_dump", "-U", db_user, "-d", db_name, "-f", backup_file ], env=env, check=True, capture_output=True ) print(f"Successfully created backup: {backup_file}") return backup_file except subprocess.CalledProcessError as e: print(f"Backup failed: {e.stderr.decode()}") return None
You’ll need to set up a Google Cloud service account first:
- Create a project in Google Cloud Console, enable the Drive API
- Create a service account, download its JSON key file
- Share your target Drive folder with the service account’s email
Here’s the upload code:
from google.oauth2.service_account import Credentials from googleapiclient.discovery import build from googleapiclient.http import MediaFileUpload def upload_to_google_drive(backup_file, drive_folder_id): # Load service account credentials creds = Credentials.from_service_account_file( "path/to/your/service_account_key.json", scopes=["https://www.googleapis.com/auth/drive"] ) service = build("drive", "v3", credentials=creds) # Define file metadata (upload to specified folder) file_metadata = { "name": os.path.basename(backup_file), "parents": [drive_folder_id] # Get this from your Drive folder URL } media = MediaFileUpload(backup_file, mimetype="application/sql") try: file = service.files().create( body=file_metadata, media_body=media, fields="id" ).execute() print(f"Uploaded to Drive with ID: {file.get('id')}") # Clean up local backup file after upload os.remove(backup_file) except Exception as e: print(f"Drive upload failed: {str(e)}")
For Dropbox, first create a Dropbox App (in the Dropbox Developer Portal) and generate an access token. Then use the official Python SDK:
import dropbox def upload_to_dropbox(backup_file, dropbox_path): # Initialize Dropbox client with access token from env var dbx = dropbox.Dropbox(os.getenv("DROPBOX_ACCESS_TOKEN")) try: with open(backup_file, "rb") as f: # Use overwrite mode to replace files with same name (optional) dbx.files_upload(f.read(), dropbox_path, mode=dropbox.files.WriteMode("overwrite")) print(f"Uploaded to Dropbox: {dropbox_path}") os.remove(backup_file) except Exception as e: print(f"Dropbox upload failed: {str(e)}")
Note: dropbox_path should look like /backups/postgres_backup_20240520_120000.sql
You have two solid options here:
Option 1: Integrate with Flask using APScheduler
This runs the backup alongside your Flask app. Install APScheduler first: pip install apscheduler
from flask import Flask from apscheduler.schedulers.background import BackgroundScheduler app = Flask(__name__) # Initialize background scheduler scheduler = BackgroundScheduler() def daily_backup_task(): backup_file = create_postgres_backup() if backup_file: # Choose your cloud storage here upload_to_google_drive(backup_file, "your_drive_folder_id") # OR upload_to_dropbox(backup_file, "/backups/" + os.path.basename(backup_file)) # Schedule task to run every day at 2 AM scheduler.add_job(daily_backup_task, "cron", hour=2) @app.route("/") def home(): return "Flask app running with automated backups!" if __name__ == "__main__": scheduler.start() app.run(debug=False) # Disable debug in production!
Option 2: Use System Cron (Linux)
If you prefer not to tie backups to your Flask process, create a standalone script (e.g., backup_script.py) with the backup and upload logic, then set up a cron job:
- Open crontab:
crontab -e - Add this line to run at 2 AM daily:
0 2 * * * /usr/bin/python3 /path/to/your/backup_script.py
Make sure the script has access to all required environment variables and permissions.
- Validate Backups: Regularly download a backup and test restoring it to a test database—don’t assume backups work until you verify!
- Clean Up Old Files: Set up cloud storage rules to delete backups older than 30-90 days (both services support this via their web interfaces).
- Secure Credentials: Store all sensitive data (DB password, cloud tokens) in environment variables or a secure secrets manager—never hardcode them!
- Check Logs: Add logging to your script (instead of print statements) so you can debug failures easily.
内容的提问来源于stack exchange,提问作者Tri




