Python脚本如何操作数据库DDL变更

本文目录导读：

1. 使用原生数据库驱动
2. 使用ORM框架 (SQLAlchemy)
3. 使用数据库迁移工具
4. 通用DDL执行器
5. 最佳实践建议

我来详细介绍Python操作数据库DDL变更的几种常见方式：

使用原生数据库驱动

MySQL (使用mysql-connector-python)

import mysql.connector
from mysql.connector import errorcode
def execute_ddl():
    config = {
        'user': 'your_user',
        'password': 'your_password',
        'host': 'localhost',
        'database': 'your_db',
        'raise_on_warnings': True
    }
    try:
        conn = mysql.connector.connect(**config)
        cursor = conn.cursor()
        # DDL 操作示例
        ddl_statements = [
            # 创建表
            """CREATE TABLE IF NOT EXISTS users (
                id INT AUTO_INCREMENT PRIMARY KEY,
                username VARCHAR(50) NOT NULL,
                email VARCHAR(100),
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )""",
            # 添加列
            "ALTER TABLE users ADD COLUMN phone VARCHAR(20) AFTER email",
            # 修改列
            "ALTER TABLE users MODIFY COLUMN phone VARCHAR(30)",
            # 添加索引
            "CREATE INDEX idx_username ON users(username)",
            # 添加外键
            """ALTER TABLE orders 
               ADD CONSTRAINT fk_user_id 
               FOREIGN KEY (user_id) REFERENCES users(id)"""
        ]
        for statement in ddl_statements:
            try:
                cursor.execute(statement)
                print(f"执行成功: {statement[:50]}...")
            except mysql.connector.Error as err:
                print(f"执行失败: {err}")
                conn.rollback()
        conn.commit()
    except mysql.connector.Error as err:
        if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
            print("用户名或密码错误")
        elif err.errno == errorcode.ER_BAD_DB_ERROR:
            print("数据库不存在")
        else:
            print(f"连接错误: {err}")
    finally:
        if 'conn' in locals() and conn.is_connected():
            cursor.close()
            conn.close()

PostgreSQL (使用psycopg2)

import psycopg2
from psycopg2 import sql
def postgresql_ddl():
    conn = psycopg2.connect(
        host="localhost",
        database="your_db",
        user="your_user",
        password="your_password"
    )
    try:
        cur = conn.cursor()
        # 使用参数化DDL
        table_name = "employees"
        column_name = "salary"
        column_type = "DECIMAL(10,2)"
        # 创建表
        create_table = sql.SQL("""
            CREATE TABLE IF NOT EXISTS {} (
                id SERIAL PRIMARY KEY,
                name VARCHAR(100) NOT NULL,
                {} {},
                hire_date DATE
            )
        """).format(
            sql.Identifier(table_name),
            sql.Identifier(column_name),
            sql.SQL(column_type)
        )
        cur.execute(create_table)
        # 检查表是否存在
        cur.execute("""
            SELECT EXISTS (
                SELECT FROM information_schema.tables 
                WHERE table_name = %s
            )
        """, (table_name,))
        exists = cur.fetchone()[0]
        print(f"表 {table_name} 是否存在: {exists}")
        conn.commit()
    except Exception as e:
        print(f"DDL执行失败: {e}")
        conn.rollback()
    finally:
        cur.close()
        conn.close()

使用ORM框架 (SQLAlchemy)

基本DDL操作

from sqlalchemy import create_engine, text, MetaData, Table, Column, Integer, String
from sqlalchemy.exc import SQLAlchemyError
def sqlalchemy_ddl():
    # 创建数据库引擎
    engine = create_engine('mysql+pymysql://user:password@localhost/dbname')
    try:
        with engine.connect() as conn:
            # 直接执行DDL
            conn.execute(text("""
                CREATE TABLE IF NOT EXISTS products (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    name VARCHAR(200) NOT NULL,
                    price DECIMAL(10,2),
                    INDEX idx_name (name)
                )
            """))
            # 使用DDL对象
            from sqlalchemy.schema import CreateTable, AddColumn
            # 添加列
            conn.execute(text("ALTER TABLE products ADD COLUMN category VARCHAR(100)"))
            # 修改列
            conn.execute(text("ALTER TABLE products MODIFY COLUMN price DECIMAL(12,2)"))
            conn.commit()
            print("DDL操作完成")
    except SQLAlchemyError as e:
        print(f"数据库错误: {e}")

使用Alembic管理DDL变更

# alembic/versions/xxxx_create_tables.py
"""create tables
Revision ID: 1234567890
Revises: 
Create Date: 2024-01-01 10:00:00.000000
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '1234567890'
down_revision = None
branch_labels = None
depends_on = None
def upgrade():
    # 创建表
    op.create_table(
        'accounts',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('name', sa.String(length=100), nullable=True),
        sa.Column('balance', sa.Float(), nullable=True),
        sa.PrimaryKeyConstraint('id')
    )
    # 添加列
    op.add_column('accounts', sa.Column('email', sa.String(length=200)))
    # 创建索引
    op.create_index('idx_account_name', 'accounts', ['name'])
def downgrade():
    # 回滚操作
    op.drop_index('idx_account_name', table_name='accounts')
    op.drop_column('accounts', 'email')
    op.drop_table('accounts')

使用数据库迁移工具

yoyo-migrations 示例

# migrations/0001_create_users.py
from yoyo import step
step(
    """
    CREATE TABLE users (
        id INT AUTO_INCREMENT PRIMARY KEY,
        username VARCHAR(50) NOT NULL UNIQUE,
        email VARCHAR(100),
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
    """,
    """
    DROP TABLE IF EXISTS users
    """
)
step(
    """
    ALTER TABLE users ADD COLUMN phone VARCHAR(20)
    """,
    """
    ALTER TABLE users DROP COLUMN phone
    """
)

通用DDL执行器

import logging
from typing import List, Callable
from contextlib import contextmanager
class DDLExecutor:
    def __init__(self, connection_config: dict):
        self.config = connection_config
        self.setup_logging()
    def setup_logging(self):
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(__name__)
    @contextmanager
    def get_connection(self):
        """获取数据库连接"""
        conn = mysql.connector.connect(**self.config)
        try:
            yield conn
        finally:
            conn.close()
    def execute_ddl(self, statements: List[str], 
                   transaction: bool = True,
                   on_error: str = 'stop'):
        """
        执行DDL语句列表
        Args:
            statements: DDL语句列表
            transaction: 是否使用事务
            on_error: 'stop' 或 'continue'
        """
        with self.get_connection() as conn:
            cursor = conn.cursor()
            try:
                for i, statement in enumerate(statements, 1):
                    try:
                        self.logger.info(f"执行DDL [{i}]: {statement[:100]}")
                        cursor.execute(statement)
                        if not transaction:
                            conn.commit()
                    except Exception as e:
                        self.logger.error(f"DDL [{i}] 失败: {e}")
                        if on_error == 'stop':
                            if transaction:
                                conn.rollback()
                            raise
                        elif on_error == 'continue':
                            continue
                if transaction:
                    conn.commit()
                    self.logger.info("所有DDL执行完成")
            finally:
                cursor.close()
    def validate_ddl(self, statement: str) -> bool:
        """验证DDL语句语法"""
        import sqlparse
        try:
            parsed = sqlparse.parse(statement)
            return len(parsed) > 0
        except:
            return False
# 使用示例
executor = DDLExecutor({
    'host': 'localhost',
    'database': 'test_db',
    'user': 'root',
    'password': 'password'
})
# 批量执行DDL
ddls = [
    "CREATE TABLE IF NOT EXISTS table1 (id INT PRIMARY KEY)",
    "ALTER TABLE table1 ADD COLUMN name VARCHAR(50)",
    "CREATE INDEX idx_name ON table1(name)"
]
try:
    executor.execute_ddl(ddls, transaction=True, on_error='stop')
except Exception as e:
    print(f"DDL执行过程出错: {e}")

最佳实践建议

安全执行DDL的封装

import time
from functools import wraps
def retry_on_lock(max_retries=3, delay=1):
    """DDL锁等待重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if "lock" in str(e).lower() and attempt < max_retries - 1:
                        time.sleep(delay * (attempt + 1))
                        continue
                    raise
            return func(*args, **kwargs)
        return wrapper
    return decorator
class SafeDDLExecutor:
    def __init__(self, connection):
        self.conn = connection
        self.backup_statements = []
    @retry_on_lock(max_retries=3)
    def execute_alter_table(self, table_name: str, alter_clause: str):
        """安全执行ALTER TABLE"""
        # 先检查表是否存在
        cursor = self.conn.cursor()
        cursor.execute(f"SHOW TABLES LIKE '{table_name}'")
        if not cursor.fetchone():
            raise ValueError(f"表 {table_name} 不存在")
        # 备份当前结构
        cursor.execute(f"SHOW CREATE TABLE {table_name}")
        create_stmt = cursor.fetchone()[1]
        self.backup_statements.append(create_stmt)
        # 执行ALTER
        alter_stmt = f"ALTER TABLE {table_name} {alter_clause}"
        cursor.execute(alter_stmt)
        self.conn.commit()
        print(f"成功执行: {alter_stmt}")
    def rollback_last(self):
        """回滚最后一次DDL变更"""
        if self.backup_statements:
            last_backup = self.backup_statements.pop()
            # 使用备份的CREATE语句重新创建表
            cursor = self.conn.cursor()
            cursor.execute(last_backup)
            self.conn.commit()
# 使用示例
conn = mysql.connector.connect(**config)
executor = SafeDDLExecutor(conn)
try:
    executor.execute_alter_table("users", "ADD COLUMN age INT DEFAULT 0")
except Exception as e:
    print(f"DDL失败: {e}")
    executor.rollback_last()
    print("已回滚变更")

1. 简单项目：直接使用原生驱动执行DDL即可
2. 复杂项目：推荐使用SQLAlchemy + Alembic管理
3. 生产环境：必须实现错误处理、事务管理和回滚机制
4. 大型变更：考虑使用在线DDL工具（如pt-online-schema-change）

选择哪种方式取决于你的项目规模、团队习惯和数据库类型，建议在生产环境操作前先在测试环境验证。