如何用脚本批量创建PostgreSQL库？

本文目录导读：

1. 方法1：使用Shell脚本
2. 方法2：使用Python脚本
3. 方法3：使用SQL脚本文件
4. 方法4：使用动态SQL生成
5. 使用示例
6. 注意事项

我来介绍几种批量创建PostgreSQL数据库的脚本方法：

方法1：使用Shell脚本

基础脚本

#!/bin/bash
# 数据库列表
databases=("db1" "db2" "db3" "db4")
# 连接参数
PGHOST="localhost"
PGPORT="5432"
PGUSER="postgres"
PGPASSWORD="your_password"
# 设置密码（注意：有安全风险）
export PGPASSWORD=$PGPASSWORD
# 批量创建数据库
for db in "${databases[@]}"; do
    echo "Creating database: $db"
    createdb -h $PGHOST -p $PGPORT -U $PGUSER $db
    if [ $? -eq 0 ]; then
        echo "✓ Database $db created successfully"
    else
        echo "✗ Failed to create database $db"
    fi
done
# 清除密码变量
unset PGPASSWORD

带参数的高级脚本

#!/bin/bash
# 用法说明
usage() {
    echo "Usage: $0 [-h host] [-p port] [-U user] [-f file] [-l list]"
    echo "  -h: PostgreSQL host (default: localhost)"
    echo "  -p: PostgreSQL port (default: 5432)"
    echo "  -U: PostgreSQL user (default: postgres)"
    echo "  -f: File containing database names (one per line)"
    echo "  -l: Comma-separated list of database names"
    echo "  -o: Owner of databases"
    echo "  -e: Encoding (default: UTF8)"
    exit 1
}
# 默认值
HOST="localhost"
PORT="5432"
USER="postgres"
ENCODING="UTF8"
OWNER=""
# 解析参数
while getopts "h:p:U:f:l:o:e:" opt; do
    case $opt in
        h) HOST="$OPTARG" ;;
        p) PORT="$OPTARG" ;;
        U) USER="$OPTARG" ;;
        f) DB_FILE="$OPTARG" ;;
        l) IFS=',' read -ra DATABASES <<< "$OPTARG" ;;
        o) OWNER="$OPTARG" ;;
        e) ENCODING="$OPTARG" ;;
        *) usage ;;
    esac
done
# 读取文件中的数据库名
if [ -n "$DB_FILE" ]; then
    if [ -f "$DB_FILE" ]; then
        mapfile -t DATABASES < "$DB_FILE"
    else
        echo "Error: File $DB_FILE not found"
        exit 1
    fi
fi
# 检查参数
if [ ${#DATABASES[@]} -eq 0 ]; then
    echo "Error: No databases specified"
    usage
fi
# 提示输入密码
read -s -p "Enter PostgreSQL password: " PGPASSWORD
echo
export PGPASSWORD
# 批量创建
SUCCESS=0
FAILED=0
for db in "${DATABASES[@]}"; do
    # 去除空格
    db=$(echo "$db" | xargs)
    if [ -z "$db" ]; then
        continue
    fi
    echo "Creating database: $db"
    # 构建SQL
    SQL="CREATE DATABASE \"$db\" ENCODING '$ENCODING'"
    if [ -n "$OWNER" ]; then
        SQL="$SQL OWNER \"$OWNER\""
    fi
    # 执行SQL
    psql -h $HOST -p $PORT -U $USER -c "$SQL" 2>&1
    if [ $? -eq 0 ]; then
        echo "✓ Database '$db' created successfully"
        ((SUCCESS++))
    else
        echo "✗ Failed to create database '$db'"
        ((FAILED++))
    fi
done
# 统计信息
echo "--------------------------------"
echo "Summary:"
echo "  Total: $((SUCCESS + FAILED))"
echo "  Success: $SUCCESS"
echo "  Failed: $FAILED"
unset PGPASSWORD

方法2：使用Python脚本

基础Python脚本

#!/usr/bin/env python3
import psycopg2
import sys
def create_databases(db_list, host='localhost', port=5432, user='postgres', password=''):
    """批量创建数据库"""
    # 连接到默认数据库
    conn = psycopg2.connect(
        host=host,
        port=port,
        user=user,
        password=password,
        dbname='postgres'
    )
    conn.autocommit = True
    cursor = conn.cursor()
    results = []
    for db_name in db_list:
        try:
            # 检查数据库是否存在
            cursor.execute(f"SELECT 1 FROM pg_database WHERE datname = '{db_name}'")
            exists = cursor.fetchone()
            if exists:
                print(f"⚠ Database '{db_name}' already exists, skipping...")
                results.append({'name': db_name, 'status': 'skipped'})
                continue
            # 创建数据库
            cursor.execute(f"CREATE DATABASE \"{db_name}\"")
            print(f"✓ Database '{db_name}' created successfully")
            results.append({'name': db_name, 'status': 'created'})
        except Exception as e:
            print(f"✗ Failed to create database '{db_name}': {e}")
            results.append({'name': db_name, 'status': 'failed', 'error': str(e)})
    cursor.close()
    conn.close()
    return results
# 使用示例
if __name__ == "__main__":
    # 数据库列表
    databases = [
        "project_a_dev",
        "project_a_test",
        "project_b_dev", 
        "project_b_test",
        "analytics_db"
    ]
    # 创建数据库
    results = create_databases(
        db_list=databases,
        host='localhost',
        port=5432,
        user='postgres',
        password='your_password'
    )
    # 打印统计信息
    print("\nSummary:")
    for r in results:
        print(f"  {r['name']}: {r['status']}")

高级Python脚本（带参数和控制）

#!/usr/bin/env python3
"""
批量创建PostgreSQL数据库的工具
"""
import argparse
import csv
import sys
import psycopg2
from psycopg2 import sql
class DatabaseCreator:
    def __init__(self, host='localhost', port=5432, user='postgres', password=''):
        self.conn_params = {
            'host': host,
            'port': port,
            'user': user,
            'password': password,
            'dbname': 'postgres'
        }
        self.conn = None
        self.cursor = None
    def connect(self):
        """建立数据库连接"""
        try:
            self.conn = psycopg2.connect(**self.conn_params)
            self.conn.autocommit = True
            self.cursor = self.conn.cursor()
            return True
        except Exception as e:
            print(f"Connection failed: {e}")
            return False
    def close(self):
        """关闭连接"""
        if self.cursor:
            self.cursor.close()
        if self.conn:
            self.conn.close()
    def database_exists(self, db_name):
        """检查数据库是否存在"""
        self.cursor.execute(
            "SELECT 1 FROM pg_database WHERE datname = %s",
            (db_name,)
        )
        return self.cursor.fetchone() is not None
    def create_database(self, db_name, owner=None, encoding='UTF8', 
                       tablespace=None, connection_limit=-1):
        """创建单个数据库"""
        try:
            if self.database_exists(db_name):
                return {'name': db_name, 'status': 'skipped', 
                       'message': 'Already exists'}
            # 构建SQL语句
            sql_parts = [sql.SQL("CREATE DATABASE {}").format(
                sql.Identifier(db_name))]
            if encoding:
                sql_parts.append(sql.SQL("ENCODING {}").format(
                    sql.Literal(encoding)))
            if owner:
                sql_parts.append(sql.SQL("OWNER {}").format(
                    sql.Identifier(owner)))
            if tablespace:
                sql_parts.append(sql.SQL("TABLESPACE {}").format(
                    sql.Identifier(tablespace)))
            if connection_limit > 0:
                sql_parts.append(sql.SQL("CONNECTION LIMIT {}").format(
                    sql.Literal(connection_limit)))
            query = sql.SQL(" ").join(sql_parts)
            self.cursor.execute(query)
            return {'name': db_name, 'status': 'created', 'message': 'Success'}
        except Exception as e:
            return {'name': db_name, 'status': 'failed', 'message': str(e)}
    def create_databases_from_list(self, db_configs):
        """从配置列表创建数据库"""
        results = []
        for config in db_configs:
            db_name = config.get('name', config if isinstance(config, str) else '')
            owner = config.get('owner', None) if isinstance(config, dict) else None
            encoding = config.get('encoding', 'UTF8') if isinstance(config, dict) else 'UTF8'
            result = self.create_database(db_name, owner, encoding)
            results.append(result)
            if result['status'] == 'created':
                print(f"✓ Created: {db_name}")
            elif result['status'] == 'skipped':
                print(f"⚠ Skipped: {db_name} ({result['message']})")
            else:
                print(f"✗ Failed: {db_name} ({result['message']})")
        return results
    def create_from_csv(self, csv_file):
        """从CSV文件读取配置"""
        db_configs = []
        try:
            with open(csv_file, 'r', encoding='utf-8') as f:
                reader = csv.DictReader(f)
                for row in reader:
                    db_configs.append(row)
            return self.create_databases_from_list(db_configs)
        except Exception as e:
            print(f"Error reading CSV file: {e}")
            return []
def main():
    parser = argparse.ArgumentParser(description='批量创建PostgreSQL数据库')
    parser.add_argument('-H', '--host', default='localhost', help='数据库主机')
    parser.add_argument('-p', '--port', type=int, default=5432, help='端口')
    parser.add_argument('-U', '--user', default='postgres', help='用户名')
    parser.add_argument('-w', '--password', help='密码')
    parser.add_argument('-d', '--databases', nargs='+', help='数据库名称列表')
    parser.add_argument('-f', '--file', help='包含数据库名称的文件（每行一个）')
    parser.add_argument('-c', '--csv', help='CSV配置文件')
    args = parser.parse_args()
    # 获取密码
    password = args.password
    if not password:
        import getpass
        password = getpass.getpass("Enter password: ")
    # 创建数据库创建器
    creator = DatabaseCreator(args.host, args.port, args.user, password)
    if not creator.connect():
        sys.exit(1)
    try:
        if args.csv:
            # 从CSV文件创建
            results = creator.create_from_csv(args.csv)
        elif args.databases:
            # 从命令行参数创建
            results = creator.create_databases_from_list(args.databases)
        elif args.file:
            # 从文件读取
            with open(args.file, 'r') as f:
                databases = [line.strip() for line in f if line.strip()]
            results = creator.create_databases_from_list(databases)
        else:
            parser.print_help()
            sys.exit(1)
        # 打印统计信息
        print("\n" + "="*50)
        print("Summary:")
        stats = {'created': 0, 'skipped': 0, 'failed': 0}
        for r in results:
            stats[r['status']] += 1
        print(f"  Created: {stats['created']}")
        print(f"  Skipped: {stats['skipped']}")
        print(f"  Failed: {stats['failed']}")
    finally:
        creator.close()
if __name__ == "__main__":
    main()

方法3：使用SQL脚本文件

创建SQL脚本文件 create_databases.sql

-- 创建多个数据库
CREATE DATABASE db1 WITH ENCODING 'UTF8' LC_COLLATE 'en_US.UTF-8' LC_CTYPE 'en_US.UTF-8';
CREATE DATABASE db2 WITH ENCODING 'UTF8' LC_COLLATE 'en_US.UTF-8' LC_CTYPE 'en_US.UTF-8';
CREATE DATABASE db3 WITH ENCODING 'UTF8' LC_COLLATE 'en_US.UTF-8' LC_CTYPE 'en_US.UTF-8';
-- 或者创建并设置所有者
CREATE DATABASE project_a OWNER user1 ENCODING 'UTF8';
CREATE DATABASE project_b OWNER user2 ENCODING 'UTF8';

执行：

psql -U postgres -f create_databases.sql

方法4：使用动态SQL生成

#!/bin/bash
# 动态生成SQL并执行
databases=("app1" "app2" "app3" "app4")
sql=""
for db in "${databases[@]}"; do
    sql+="CREATE DATABASE \"$db\" ENCODING 'UTF8'; "
done
# 一次性执行所有SQL
psql -U postgres -c "$sql"

使用示例

使用Shell脚本

# 基本用法
chmod +x create_dbs.sh
./create_dbs.sh
# 使用文件
echo -e "db1\ndb2\ndb3" > databases.txt
./create_dbs.sh -f databases.txt
# 指定参数
./create_dbs.sh -h dbhost -p 5432 -U admin -l "db1,db2,db3"

使用Python脚本

# 基本用法
python3 create_databases.py -d db1 db2 db3
# 使用文件
python3 create_databases.py -f databases.txt
# 使用CSV配置
python3 create_databases.py -c db_config.csv

CSV文件格式示例 (db_config.csv):

name,owner,encoding
project_a_dev,dev_user,UTF8
project_b_test,test_user,UTF8
analytics,analyst,UTF8

注意事项

1. 权限问题：需要具有创建数据库的权限（通常是superuser或createdb权限）
2. 密码安全：避免在命令行中直接输入密码，使用环境变量或交互式输入
3. 错误处理：适当的错误处理和日志记录
4. 重复检查：创建前检查数据库是否已存在
5. 字符编码：注意设置正确的编码和排序规则

这些脚本都可以根据你的具体需求进行修改和扩展。