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 ###