Python sqlalchemy 模块,SmallInteger() 实例源码
我们从Python开源项目中,提取了以下21个代码示例,用于说明如何使用sqlalchemy.SmallInteger()。
def _create_post_table(self):
"""
Creates the table to store the blog posts.
:return:
"""
with self._engine.begin() as conn:
post_table_name = self._table_name("post")
if not conn.dialect.has_table(conn, post_table_name):
self._post_table = sqla.Table(
post_table_name, self._Metadata,
sqla.Column("id", sqla.Integer, primary_key=True),
sqla.Column("title", sqla.String(256)),
sqla.Column("text", sqla.Text),
sqla.Column("post_date", sqla.DateTime),
sqla.Column("last_modified_date",
# if 1 then make it a draft
sqla.Column("draft", sqla.SmallInteger, default=0)
)
self._logger.debug("Created table with table name %s" %
post_table_name)
else:
self._post_table = self._Metadata.tables[post_table_name]
self._logger.debug("Reflecting to table with table name %s" %
post_table_name)
def _create_post_table(self):
"""
Creates the table to store the blog posts.
:return:
"""
with self._engine.begin() as conn:
post_table_name = self._table_name("post")
if not conn.dialect.has_table(conn, default=0)
)
self._logger.debug("Created table with table name %s" %
post_table_name)
else:
self._post_table = self._Metadata.tables[post_table_name]
self._logger.debug("Reflecting to table with table name %s" %
post_table_name)
def _create_post_table(self):
"""
Creates the table to store the blog posts.
:return:
"""
with self._engine.begin() as conn:
post_table_name = self._table_name("post")
if not conn.dialect.has_table(conn, default=0)
)
self._logger.debug("Created table with table name %s" %
post_table_name)
else:
self._post_table = self._Metadata.tables[post_table_name]
self._logger.debug("Reflecting to table with table name %s" %
post_table_name)
def create_table(self):
with self._engine.begin() as conn:
if not conn.dialect.has_table(conn, self.table_name):
self._count_table = sqla.Table(
self.table_name,
sqla.Column("page_id", sqla.String(256), sqla.String(512)),
sqla.Column("count", default=0),
sqla.Column("deleted", sqla.DateTime)
)
self._logger.debug("Created table with table name %s" %
self.table_name)
else:
self._count_table = self._Metadata.tables[self.table_name]
self._logger.debug("Reflecting to table with table name %s" %
self.table_name)
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('roles',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(length=64), nullable=True),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('name')
)
op.create_table('users',
sa.Column('number', sa.SmallInteger(),
sa.Column('username',
sa.Column('password', sa.String(length=128),
sa.Column('role_id',
sa.ForeignKeyConstraint(['role_id'], ['roles.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_users_number'), 'users', ['number'], unique=True)
op.create_index(op.f('ix_users_username'), ['username'], unique=False)
# ### end Alembic commands ###
def upgrade(pyramid_env):
with context.begin_transaction():
op.add_column(
'idea', sa.Column(
'hidden', sa.SmallInteger, server_default='0'))
op.execute('UPDATE idea set hidden=0')
op.execute('ALTER TABLE idea ADD CHECK (hidden IN (0,1))')
op.add_column(
'idea', sa.Column(
'widget_id', sa.Integer, sa.ForeignKey('widget.id')))
op.add_column(
'content', server_default='0'))
op.execute('UPDATE content set hidden=0')
op.execute('ALTER TABLE content ADD CHECK (hidden IN (0,1))')
# Do stuff with the app's models here.
from assembl import models as m
db = m.get_session_maker()()
with transaction.manager:
pass
def downgrade(pyramid_env):
with context.begin_transaction():
op.add_column(
'agent_email_account',
sa.Column("preferred",
default=False, server_default='0'))
# Do stuff with the app's models here.
from assembl import models as m
db = m.get_session_maker()()
with transaction.manager:
# get from prevIoUs values
db.execute("""UPDATE agent_email_account SET preferred=(
SELECT abstract_agent_account.preferred
FROM abstract_agent_account
WHERE abstract_agent_account.id = agent_email_account.id
AND abstract_agent_account."type" = 'agent_email_account')""")
# Force update,transaction manager saw nothing
aaa = db.query(m.Role).first()
flag_modified(aaa, 'name')
with context.begin_transaction():
db.execute('ALTER TABLE agent_email_account ADD CHECK (preferred IN (0,1))')
op.drop_column(
'abstract_agent_account', "preferred")
def downgrade(pyramid_env):
schema, user = config.get('db_schema'), config.get('db_user')
with context.begin_transaction():
for tablename in tables:
index_name = "%s_%s_%s_UNQC_base_id_tombstone_date" % (
schema, user, tablename)
op.drop_constraint(index_name, tablename, schema=schema)
op.add_column(tablename, sa.Column('is_tombstone', server_default="0"))
# repopulate is_tombstone
with transaction.manager:
for tablename in tables:
op.execute('UPDATE %s set "is_tombstone" = (CASE WHEN "tombstone_date" IS NULL THEN 0 ELSE 1 END)' % (tablename,))
mark_changed()
with context.begin_transaction():
for tablename in tables:
op.drop_column(tablename, "base_id")
op.drop_column(tablename, "tombstone_date")
op.drop_table(tablename+ID_TABLE)
op.execute('ALTER TABLE %s ADD CHECK ("is_tombstone" IN (0,1))'%(tablename,))
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.create_table('component_update',
sa.Column('component',
sa.Column('label',
sa.Column('status',
sa.Column('status_description', sa.Text(),
sa.Column('health',
sa.Column('tags',
sa.Column('time',
sa.Column('lifetime',
sa.PrimaryKeyConstraint('id')
)
### end Alembic commands ###
def upgrade(pyramid_env):
with context.begin_transaction():
op.add_column(
"langstring_entry",
sa.Column("error_code", sa.SmallInteger))
def upgrade(pyramid_env):
with context.begin_transaction():
op.add_column("content_source",
sa.Column("connection_error", sa.SmallInteger))
op.add_column("content_source",
sa.Column("error_description", sa.String))
def upgrade(pyramid_env):
with context.begin_transaction():
op.add_column(
'abstract_agent_account', server_default='0'))
# Do stuff with the app's models here.
from assembl import models as m
db = m.get_session_maker()()
with transaction.manager:
# Start with a blanket 0
db.execute("UPDATE abstract_agent_account SET preferred=0")
# get from prevIoUs values
db.execute("""UPDATE abstract_agent_account SET preferred=(
SELECT agent_email_account.preferred
FROM agent_email_account
WHERE abstract_agent_account.id = agent_email_account.id)
WHERE abstract_agent_account."type" = 'agent_email_account'""")
# Force update, 'name')
with context.begin_transaction():
db.execute('ALTER TABLE abstract_agent_account ADD CHECK (preferred IN (0,1))')
op.drop_column('agent_email_account', "preferred")
def upgrade(pyramid_env):
with context.begin_transaction():
op.add_column(
'extract', sa.Column(
'important', server_default='0'))
def upgrade(pyramid_env):
with context.begin_transaction():
# Is it me,or is virtuoso getting stupider?
# It used to be possible to do the CHECK at the same time,like this:
# op.add_column(
# 'local_user_role',
# sa.Column('requested',sa.BOOLEAN,server_default='0'))
# Now doing it in two phases.
op.add_column(
'local_user_role',
sa.Column('requested', server_default='0'))
op.execute('UPDATE local_user_role set requested = 0')
op.execute('ALTER TABLE local_user_role ADD CHECK (requested IN (0,1))')
def upgrade(pyramid_env):
with context.begin_transaction():
op.add_column(
'idea', sa.Column(
'is_tombstone', server_default='0'))
op.execute('UPDATE idea set hidden=0')
op.execute('ALTER TABLE idea ADD CHECK (is_tombstone IN (0,1))')
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('games',
sa.Column('winner', sa.Enum('white', 'black', name='color'),
sa.Column('handicap',
sa.Column('komi',
sa.Column('season',
sa.Column('episode',
sa.PrimaryKeyConstraint('id')
)
op.create_table('players',
sa.Column('first_name', sa.String(length=30),
sa.Column('last_name',
sa.Column('aga_id',
sa.Column('aga_rank',
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_players_aga_id'), 'players', ['aga_id'], unique=True)
op.create_table('users', sa.String(length=80),
sa.Column('email', sa.Binary(),
sa.Column('created_at', sa.DateTime(),
sa.Column('active', sa.Boolean(),
sa.Column('is_admin',
sa.UniqueConstraint('email'),
sa.UniqueConstraint('username')
)
op.create_table('black_player_games',
sa.Column('player_id',
sa.Column('game_id',
sa.ForeignKeyConstraint(['game_id'], ['games.id'],
sa.ForeignKeyConstraint(['player_id'], ['players.id'],
sa.PrimaryKeyConstraint('player_id', 'game_id')
)
op.create_table('roles',
sa.Column('user_id',
sa.ForeignKeyConstraint(['user_id'], ['users.id'],
sa.UniqueConstraint('name')
)
op.create_table('white_player_games', 'game_id')
)
# ### end Alembic commands ###
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
'vehicles',
sa.Column('id',
sa.Column('name', sa.String(length=254),
sa.Column('is_active',
sa.PrimaryKeyConstraint('id', name=op.f('pk_vehicles')),
MysqL_engine='InnoDB'
)
op.create_table(
'fuellog',
sa.Column('date', sa.Date(),
sa.Column('vehicle_id',
sa.Column('odometer_miles',
sa.Column('reported_miles',
sa.Column('calculated_miles',
sa.Column('level_before',
sa.Column('level_after',
sa.Column('fill_location',
sa.Column(
'cost_per_gallon',
sa.Numeric(precision=10, scale=4),
nullable=True
),
sa.Column(
'total_cost',
sa.Column(
'gallons',
sa.Column(
'reported_mpg',
sa.Column(
'calculated_mpg',
sa.Column('notes',
sa.ForeignKeyConstraint(
['vehicle_id'],
['vehicles.id'],
name=op.f('fk_fuellog_vehicle_id_vehicles')
), name=op.f('pk_fuellog')),
MysqL_engine='InnoDB'
)
# ### end Alembic commands ###
def upgrade():
'''
upgrade method
'''
create_table('rules_hist_recent',
sa.Column('history_id', GUID()),
sa.Column('id',
sa.Column('subscription_id',
sa.Column('account', sa.String(25)),
sa.Column('scope',
sa.Column('name', sa.String(255)),
sa.Column('did_type', DIDType.db_type()),
sa.Column('state', RuleState.db_type()),
sa.Column('error',
sa.Column('rse_expression',
sa.Column('copies', sa.SmallInteger),
sa.Column('expires_at', sa.DateTime),
sa.Column('weight',
sa.Column('locked', sa.Boolean()),
sa.Column('locks_ok_cnt', sa.BigInteger),
sa.Column('locks_replicating_cnt',
sa.Column('locks_stuck_cnt',
sa.Column('source_replica_expression',
sa.Column('activity', sa.String(50)),
sa.Column('grouping', RuleGrouping.db_type()),
sa.Column('notification', RuleNotification.db_type()),
sa.Column('stuck_at',
sa.Column('purge_replicas',
sa.Column('ignore_availability',
sa.Column('updated_at',
sa.Column('created_at', sa.DateTime))
create_table('rules_history', sa.DateTime))
if context.get_context().dialect.name != 'sqlite':
create_index('RULES_HIST_RECENT_ID_IDX', 'rules_hist_recent', ["id"])
def upgrade(pyramid_env):
with context.begin_transaction():
op.add_column(
'abstract_agent_account',
sa.Column("verified", server_default='0'))
op.add_column(
'abstract_agent_account',
sa.Column('email', sa.String(100), index=True))
# Do stuff with the app's models here.
from assembl import models as m
db = m.get_session_maker()()
with transaction.manager:
# Start with a blanket 0
db.execute("UPDATE abstract_agent_account SET verified=0")
# get from prevIoUs values
db.execute("""UPDATE abstract_agent_account SET email=(
SELECT agent_email_account.email
FROM agent_email_account
WHERE abstract_agent_account.id = agent_email_account.id)
WHERE abstract_agent_account."type" = 'agent_email_account'""")
db.execute("""UPDATE abstract_agent_account SET verified=(
SELECT agent_email_account.verified
FROM agent_email_account
WHERE abstract_agent_account.id = agent_email_account.id)
WHERE abstract_agent_account."type" = 'agent_email_account'""")
db.execute("""UPDATE abstract_agent_account SET verified=(
SELECT identity_provider.trust_emails
FROM identity_provider
JOIN idprovider_agent_account ON (
idprovider_agent_account.provider_id = identity_provider.id)
WHERE abstract_agent_account.id = idprovider_agent_account.id)
WHERE abstract_agent_account."type" = 'idprovider_agent_account'""")
db.flush()
ipaccounts = db.query(m.IdentityProviderAccount).all()
for ipaccount in ipaccounts:
ipaccount.interpret_profile()
if ipaccount.email:
email_accounts = db.query(m.EmailAccount).filter_by(
email=ipaccount.email).all()
for email_account in email_accounts:
if email_account.profile == ipaccount.profile:
ipaccount.verified |= email_account.verified
db.delete(email_account)
elif ipaccount.verified and not email_account.verified:
db.delete(email_account)
else:
# I checked that this case did not happen
# in our existing databases
ipaccount.profile.merge(email_account.profile)
with context.begin_transaction():
db.execute('ALTER TABLE abstract_agent_account ADD CHECK (verified IN (0,1))')
op.drop_table('agent_email_account')
def downgrade(pyramid_env):
with context.begin_transaction():
op.create_table(
'agent_email_account',
sa.Column('id', sa.ForeignKey(
'abstract_agent_account.id', ondelete='CASCADE',
onupdate='CASCADE'),
primary_key=True), nullable=False, index=True),
sa.Column('verified', server_default='0'),
sa.Column('active', server_default='1'))
# Do stuff with the app's models here.
from assembl import models as m
db = m.get_session_maker()()
with transaction.manager:
# get from prevIoUs values
db.execute("""INSERT INTO agent_email_account (id,email,verified)
SELECT abstract_agent_account.id,
abstract_agent_account.email,
abstract_agent_account.verified
FROM abstract_agent_account
WHERE abstract_agent_account.email IS NOT NULL
AND abstract_agent_account."type" = 'agent_email_account'""")
ipaccounts = db.query(m.IdentityProviderAccount).all()
for ipaccount in ipaccounts:
ipaccount.interpret_profile()
if ipaccount.email:
db.add(m.EmailAccount(
email=ipaccount.email, profile_id=ipaccount.profile_id,
verified=ipaccount.verified, preferred=ipaccount.preferred))
email_accounts = db.query(m.EmailAccount).filter_by(
email=ipaccount.email).all()
for email_account in email_accounts:
if email_account.profile == ipaccount.profile:
ipaccount.verified |= email_account.verified
db.delete(email_account)
elif ipaccount.verified and not email_account.verified:
db.delete(email_account)
else:
# I checked that this case did not happen
# in our existing databases
ipaccount.profile.merge(email_account.profile)
with context.begin_transaction():
db.execute(
"ALTER TABLE agent_email_account ADD CHECK (verified IN (0,1))")
op.drop_column(
'abstract_agent_account', "verified")
op.drop_index('ix_abstract_agent_account_email')
op.drop_column('abstract_agent_account', "email")
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。