Python sqlalchemy 模块，TIMESTAMP 实例源码
我们从Python开源项目中，提取了以下39个代码示例，用于说明如何使用sqlalchemy.TIMESTAMP。
def __init__(self, timezone=False, fsp=None):
"""Construct a MysqL TIMESTAMP type.
:param timezone: not used by the MysqL dialect.
:param fsp: fractional seconds precision value.
MysqL 5.6.4 supports storage of fractional seconds;
this parameter will be used when emitting DDL
for the TIMESTAMP type.
.. note::
DBAPI driver support for fractional seconds may
be limited; current support includes
MysqL Connector/Python.
.. versionadded:: 0.8.5 Added MysqL-specific :class:`.MysqL.TIMESTAMP`
with fractional seconds support.
"""
super(TIMESTAMP, self).__init__(timezone=timezone)
self.fsp = fsp
def visit_typeclause(self, typeclause):
type_ = typeclause.type.dialect_impl(self.dialect)
if isinstance(type_, sqltypes.Integer):
if getattr(type_, 'unsigned', False):
return 'UNSIGNED INTEGER'
else:
return 'SIGNED INTEGER'
elif isinstance(type_, sqltypes.TIMESTAMP):
return 'DATETIME'
elif isinstance(type_, (sqltypes.DECIMAL, sqltypes.DateTime,
sqltypes.Date, sqltypes.Time)):
return self.dialect.type_compiler.process(type_)
elif isinstance(type_, sqltypes.String) \
and not isinstance(type_, (ENUM, SET)):
adapted = CHAR._adapt_string_for_cast(type_)
return self.dialect.type_compiler.process(adapted)
elif isinstance(type_, sqltypes._Binary):
return 'BINARY'
elif isinstance(type_, sqltypes.NUMERIC):
return self.dialect.type_compiler.process(
type_).replace('NUMERIC', 'DECIMAL')
else:
return None
def get_column_specification(self, column, **kw):
"""Builds column DDL."""
colspec = [self.preparer.format_column(column),
self.dialect.type_compiler.process(column.type)
]
default = self.get_column_default_string(column)
if default is not None:
colspec.append('DEFAULT ' + default)
is_timestamp = isinstance(column.type, sqltypes.TIMESTAMP)
if not column.nullable and not is_timestamp:
colspec.append('NOT NULL')
# see: http://docs.sqlalchemy.org/en/latest/dialects/
# MysqL.html#MysqL_timestamp_null
elif column.nullable and is_timestamp and default is None:
colspec.append('NULL')
if column is column.table._autoincrement_column and \
column.server_default is None:
colspec.append('AUTO_INCREMENT')
return ' '.join(colspec)
def get_column_specification(self, sqltypes.TIMESTAMP)
if not column.nullable and not is_timestamp:
colspec.append('NOT NULL')
# see: http://docs.sqlalchemy.org/en/latest/dialects/
# MysqL.html#MysqL_timestamp_null
elif column.nullable and is_timestamp and default is None:
colspec.append('NULL')
if column is column.table._autoincrement_column and \
column.server_default is None:
colspec.append('AUTO_INCREMENT')
return ' '.join(colspec)
def downgrade():
### commands auto generated by Alembic - please adjust! ###
op.alter_column(u'users', 'user_id',
existing_type=sa.VARCHAR(length=100),
nullable=True)
op.alter_column(u'users', 'updated_at',
existing_type=postgresql.TIMESTAMP(), 'name', 'last_name',
existing_type=sa.VARCHAR(length=200), 'email',
existing_type=sa.VARCHAR(length=150), 'created_at',
nullable=True)
op.drop_table('request_referral')
### end Alembic commands ###
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('email',
sa.Column('user_id', zeus.db.types.guid.GUID(),
nullable=False),
sa.Column('email', sa.String(length=128), nullable=False),
sa.Column('verified', sa.Boolean(), nullable=True),
sa.Column('id',
sa.Column('date_created', sa.TIMESTAMP(timezone=True),
server_default=sa.text('Now()'),
sa.ForeignKeyConstraint(
['user_id'], ['user.id'], ondelete='CASCADE'),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('user_id',
name='unq_user_email')
)
op.create_index(op.f('ix_email_user_id'),
['user_id'], unique=False)
# ### end Alembic commands ###
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
'failurereason',
sa.Column('job_id',
sa.Column('reason', zeus.db.types.enum.StrEnum(),
sa.Column('repository_id',
sa.Column('id',
sa.Column(
'date_created',
sa.TIMESTAMP(timezone=True),
server_default=sa.text('Now()'),
nullable=False
),
sa.ForeignKeyConstraint(['job_id'], ['job.id'],
sa.ForeignKeyConstraint(['repository_id'], ['repository.id'],
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('job_id', 'reason', name='unq_failurereason_key')
)
op.create_index(
op.f('ix_failurereason_repository_id'), 'failurereason', ['repository_id'], unique=False
)
# ### end Alembic commands ###
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('web_hook',
sa.Column('id', postgresql.UUID(as_uuid=True),
sa.Column('name', sa.TEXT(), nullable=False, unique=True),
sa.Column('description',
sa.Column('url',
sa.Column('shared_secret',
sa.Column('status', sa.Integer(),
sa.Column('consecutive_failure_count',
sa.Column('register_time', sa.TIMESTAMP(),
sa.Column('created_by_uid',
sa.Column('permissions', server_default='"[]"'),
sa.PrimaryKeyConstraint('id')
)
op.create_table('web_hook_token',
sa.Column('token_id',
sa.Column('user_id',
sa.Column('web_hook_id',
sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
sa.ForeignKeyConstraint(['web_hook_id'], ['web_hook.id'],
sa.PrimaryKeyConstraint('user_id', 'web_hook_id')
)
# ### end Alembic commands ###
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.create_table('favorites',
sa.Column('bangumi_id',
sa.ForeignKeyConstraint(['bangumi_id'], ['bangumi.id'],
sa.PrimaryKeyConstraint('id')
)
op.create_table('watch_progress',
sa.Column('episode_id',
sa.Column('watch_status',
sa.Column('last_watch_position',
sa.Column('last_watch_time',
sa.ForeignKeyConstraint(['episode_id'], ['episodes.id'],
sa.PrimaryKeyConstraint('id')
)
### end Alembic commands ###
def visit_TIMESTAMP(self, type_):
if getattr(type_, 'fsp', None):
return "TIMESTAMP(%d)" % type_.fsp
else:
return "TIMESTAMP"
def visit_TIMESTAMP(self, None):
return "TIMESTAMP(%d)" % type_.fsp
else:
return "TIMESTAMP"
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('event', sa.Column('created_by', sa.String(), nullable=True))
op.add_column('event', sa.Column('end_time', sa.Column('start_time', sa.Column('url', nullable=True))
# ### end Alembic commands ###
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('event', sa.Column('created_at', nullable=True))
# ### end Alembic commands ###
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.add_column('users', sa.Column('address', sa.String(length=255), nullable=True))
op.add_column('users', sa.Column('birth_date', sa.Column('last_name', sa.String(length=200), sa.Column('updated_at', sa.Column('user_id', sa.String(length=100), nullable=True))
### end Alembic commands ###
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.create_table('user_session',
sa.Column('session_start',
sa.Column('session_end',
sa.Column('isValid',
sa.Column('user_plan',
sa.Column('created_at',
sa.Column('updated_at',
sa.PrimaryKeyConstraint('id')
)
### end Alembic commands ###
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.create_table('request_referral',
sa.Column('from_user',
sa.Column('to_user',
sa.Column('request_type',
sa.Column('referral_id',
sa.PrimaryKeyConstraint('id')
)
op.alter_column(u'users',
nullable=False)
op.alter_column(u'users',
nullable=False)
### end Alembic commands ###
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column(
'artifact',
sa.Column(
'date_finished',
sa.TIMESTAMP(
timezone=True),
nullable=True))
op.add_column('artifact', sa.Column('date_started', nullable=True))
op.add_column('artifact', sa.Column('date_updated', nullable=True))
# ### end Alembic commands ###
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('job', nullable=True))
# ### end Alembic commands ###
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('styleviolation',
sa.Column('job_id',
sa.Column('filename', sa.Text(),
sa.Column('severity', zeus.db.types.enum.Enum(),
sa.Column('message',
sa.Column('lineno',
sa.Column('colno',
sa.Column('source',
sa.Column('repository_id',
sa.Column(
'date_created',
sa.TIMESTAMP(
timezone=True),
server_default=sa.text('Now()'),
nullable=False),
sa.ForeignKeyConstraint(['job_id'],
sa.ForeignKeyConstraint(
['repository_id'],
sa.PrimaryKeyConstraint('id')
)
op.create_index(
op.f('ix_styleviolation_repository_id'),
'styleviolation',
['repository_id'],
unique=False)
# ### end Alembic commands ###
def quotes_table(Metadata, market):
table_name = quote_key(market)
# Index columns
ts_col = sa.Column('Timestamp', sa.TIMESTAMP, nullable=False)
asset_col = sa.Column(
config.ASSET_ID,
sa.INTEGER,
sa.ForeignKey(config.ASSETS + '.' + config.ID),
nullable=False
)
return sa.Table(
table_name,
Metadata,
ts_col,
asset_col,
sa.Column('Bid', sa.INTEGER, # No example of nullable
sa.Column('Ask',
sa.Column('Last',
sa.Column('Low',
sa.Column('High',
sa.Column('Average', sa.DECIMAL,
sa.Index('idx', asset_col, ts_col,
)
def upgrade():
op.create_table('vpp_etcd_journal', sa.Integer, primary_key=True,
autoincrement=True,
sa.Column('k', sa.String(255),
sa.Column('v', sa.PickleType,
sa.Column('retry_count', default=0),
sa.Column('created_at', sa.DateTime,
default=sa.func.Now()),
sa.Column('last_retried',
server_default=sa.func.Now(),
onupdate=sa.func.Now()))
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('task',
sa.Column('type',
sa.Column('content',
sa.PrimaryKeyConstraint('id')
)
op.add_column(u'bangumi', sa.Column('delete_mark', nullable=True))
op.add_column(u'episodes', nullable=True))
# ### end Alembic commands ###
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.add_column('users', sa.Column('email', sa.String(length=512), unique=True, sa.Column('email_confirmed', sa.BOOLEAN(), server_default=sa.false()))
op.add_column('users', sa.Column('register_time', server_default=sa.func.Now()))
op.add_column('users', sa.Column('update_time', server_default=sa.func.Now()))
### end Alembic commands ###
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('announce',
sa.Column('image_url',
sa.Column('position',
sa.Column('sort_order',
sa.Column('start_time',
sa.Column('end_time',
sa.PrimaryKeyConstraint('id')
)
# ### end Alembic commands ###
def add_column(cls, operations, table_name, schema=None):
"""Issue an "add column" instruction using the current
migration context.
e.g.::
from alembic import op
from sqlalchemy import Column,String
op.add_column('organization',
Column('name',String())
)
The provided :class:`~sqlalchemy.schema.Column` object can also
specify a :class:`~sqlalchemy.schema.ForeignKey`,referencing
a remote table name. Alembic will automatically generate a stub
"referenced" table and emit a second ALTER statement in order
to add the constraint separately::
from alembic import op
from sqlalchemy import Column,INTEGER,ForeignKey
op.add_column('organization',
Column('account_id',ForeignKey('accounts.id'))
)
Note that this statement uses the :class:`~sqlalchemy.schema.Column`
construct as is from the sqlAlchemy library. In particular,
default values to be created on the database side are
specified using the ``server_default`` parameter,and not
``default`` which only specifies Python-side defaults::
from alembic import op
from sqlalchemy import Column,TIMESTAMP,func
# specify "DEFAULT Now" along with the column add
op.add_column('account',
Column('timestamp',server_default=func.Now())
)
:param table_name: String name of the parent table.
:param column: a :class:`sqlalchemy.schema.Column` object
representing the new column.
:param schema: Optional schema name to operate within. To control
quoting of the schema outside of the default behavior,use
the sqlAlchemy construct
:class:`~sqlalchemy.sql.elements.quoted_name`.
.. versionadded:: 0.7.0 'schema' can Now accept a
:class:`~sqlalchemy.sql.elements.quoted_name` construct.
"""
op = cls(table_name, schema=schema)
return operations.invoke(op)
def add_column(self,server_default=func.Now())
)
:param table_name: String name of the parent table.
:param column: a :class:`sqlalchemy.schema.Column` object
representing the new column.
:param schema: Optional schema name to operate within.
.. versionadded:: 0.4.0
"""
t = self._table(table_name, schema=schema)
self.impl.add_column(
table_name,
column,
schema=schema
)
for constraint in t.constraints:
if not isinstance(constraint, sa_schema.PrimaryKeyConstraint):
self.impl.add_constraint(constraint)
def add_column(cls, schema=schema)
return operations.invoke(op)
def add_column(self, sa_schema.PrimaryKeyConstraint):
self.impl.add_constraint(constraint)
def create_table(self, name, *columns, **kw):
"""Issue a "create table" instruction using the current migration context.
This directive receives an argument list similar to that of the
Traditional :class:`sqlalchemy.schema.Table` construct,but without the
Metadata::
from sqlalchemy import INTEGER,VARCHAR,NVARCHAR,Column
from alembic import op
op.create_table(
'account',
Column('id',primary_key=True),VARCHAR(50),nullable=False),
Column('description',NVARCHAR(200))
Column('timestamp',server_default=func.Now())
)
Note that :meth:`.create_table` accepts :class:`~sqlalchemy.schema.Column`
constructs directly from the sqlAlchemy library. In particular,func
# specify "DEFAULT Now" along with the "timestamp" column
op.create_table('account',server_default=func.Now())
)
:param name: Name of the table
:param \*columns: collection of :class:`~sqlalchemy.schema.Column`
objects within
the table,as well as optional :class:`~sqlalchemy.schema.Constraint`
objects
and :class:`~.sqlalchemy.schema.Index` objects.
:param schema: Optional schema name to operate within.
:param \**kw: Other keyword arguments are passed to the underlying
:class:`sqlalchemy.schema.Table` object created for the command.
"""
self.impl.create_table(
self._table(name, **kw)
)
def add_column(cls, schema=schema)
return operations.invoke(op)
def add_column(cls, schema=schema)
return operations.invoke(op)
def add_column(cls, schema=schema)
return operations.invoke(op)
def add_column(cls, schema=schema)
return operations.invoke(op)
def add_column(cls, schema=schema)
return operations.invoke(op)
def add_column(cls, schema=schema)
return operations.invoke(op)
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。
