You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

如何基于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:

Core Workflow Overview

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.

Step 1: Create PostgreSQL Backup File

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
Step 2: Upload to Google Drive

You’ll need to set up a Google Cloud service account first:

  1. Create a project in Google Cloud Console, enable the Drive API
  2. Create a service account, download its JSON key file
  3. 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)}")
Step 3: Upload to Dropbox

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

Step 4: Automate Daily Backups

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:

  1. Open crontab: crontab -e
  2. 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.

Critical Tips for Reliability
  • 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

火山引擎 最新活动