zoukankan      html  css  js  c++  java
  • 如何在Alembic upgrade脚本中执行插入和更新数据库记录 [Flask] [Sqlalchemy] [Alembic] [python]

    1. 插入数据

    """empty message
    
    Revision ID: c15fcfd8e40a
    Revises: 35ca68348a8d
    Create Date: 2021-01-05 19:39:22.768992
    
    """
    from alembic import op
    import sqlalchemy as sa
    from sqlalchemy.sql import table, column
    
    # revision identifiers, used by Alembic.
    revision = 'c15fcfd8e40a'
    down_revision = '35ca68348a8d'
    branch_labels = None
    depends_on = None
    
    
    def upgrade():
        # ### commands auto generated by Alembic - please adjust! ###
        op.create_table(
            'users',
            sa.Column('id', sa.String(length=45), nullable=False),
            sa.Column('name', sa.String(length=100), nullable=False),
            sa.PrimaryKeyConstraint('id')
        )
    
        # insert records
        user_table = table(
            'users',
            column('id', sa.String),
            column('name', sa.String)
        )
        op.bulk_insert(
            user_table,
            [
                {
                    "id": "123",
                    "name": "Hello"
                },
                {
                    "id": "234",
                    "name": "World"
                },
            ]
        )
    
    
    def downgrade():
        # ### commands auto generated by Alembic - please adjust! ###
        op.drop_table('users')
        # ### end Alembic commands ###
    
    

    2. 更新数据库中的数据

    """empty message
    
    Revision ID: c15fcfd8e40a
    Revises: 35ca68348a8d
    Create Date: 2021-01-05 19:39:22.768992
    
    """
    from alembic import op
    import sqlalchemy as sa
    
    # revision identifiers, used by Alembic.
    revision = 'c15fcfd8e40a'
    down_revision = '35ca68348a8d'
    branch_labels = None
    depends_on = None
    
    
    def upgrade():
        # ### commands auto generated by Alembic - please adjust! ###
        op.create_table(
            'users',
            sa.Column('id', sa.String(length=45), nullable=False),
            sa.Column('name', sa.String(length=100), nullable=False),
            sa.Column('password', sa.String(length=150), nullable=False),
            sa.PrimaryKeyConstraint('id')
        )
        # update records
        bind = op.get_bind()
        for user in bind.execute('select id, name from users'):
            if user[1] is None:
                bind.execute("update users set name = %s where id = '%s'" % ("hello_world", user[0]))
        # ### end Alembic commands ###
    
    def downgrade():
        # ### commands auto generated by Alembic - please adjust! ###
        op.drop_table('users')
        # ### end Alembic commands ###
    
  • 相关阅读:
    数据恢复:模拟2个逻辑坏块
    Oracle Internal Event:10201 consistent read undo application诊断事件
    Enterprise Manager Cloud Control 12.1.0.1 (12c)安装指南
    Script:GoldenGate For Oracle数据库预检查脚本
    OGG常见初始化方案
    expdp+compression性能测试
    了解AMDU工具生成的MAP文件
    何时会发生db file sequential read等待事件?
    Oracle发布公共云Public Cloud
    Oracle Internal Research深入研究Oracle内部原理
  • 原文地址:https://www.cnblogs.com/guohewei/p/14975029.html
Copyright © 2011-2022 走看看