Python sqlalchemy 模块，dialects() 实例源码
我们从Python开源项目中，提取了以下9个代码示例，用于说明如何使用sqlalchemy.dialects()。
def load_dialect_impl(self, dialect: dialects) -> DialectType:
"""
sqlAlchemy wraps all database-specific features into
dialects,which are then responsible for generating the sql code
for a specific DB type when loading in data. ``load_dialect_impl``
is called when CRUD (create,update,delete operations) needs to be
done on the database. This method is responsible for telling
sqlAlchemy how to configure the dialect to write this type
:param dialect: The loaded dialect
:return: The type descriptor for this type.
"""
if dialect.name == 'postgresql':
return dialect.type_descriptor(postgresql.JSON())
elif dialect.name == 'MysqL':
if 'JSON' in dialect.ischema_names:
return dialect.type_descriptor(MysqL.JSON())
else:
return dialect.type_descriptor(
VARCHAR(self._MAX_VARCHAR_LIMIT)
)
else:
return dialect.type_descriptor(VARCHAR(self._MAX_VARCHAR_LIMIT))
def process_bind_param(
self, value: ValueType, dialect: dialects
) -> Optional[str]:
"""
Given a value and a dialect,determine how to serialize the type to
the dialect
.. note::
python3 will complain that int is not supported for this type.
I want to ignore this if possible
:param value: The value to encode
:param dialect: The dialect to which this will be encoded to
:return: The value encoded in that dialect
"""
if value is None:
return value
else:
return json.dumps(value)
def load_dialect_impl(self,delete operations) needs to be
done on the database. This method is responsible for telling
sqlAlchemy how to configure the dialect to write this type
:param dialect: The loaded dialect
:return: The type descriptor for this type.
"""
if dialect.name == 'postgresql':
return dialect.type_descriptor(postgresql.UUID())
else:
return dialect.type_descriptor(CHAR(32))
def process_bind_param(
self,determine how to serialize the type to
the dialect
.. note::
python3 will complain that int is not supported for this type.
I want to ignore this if possible
:param value: The value to encode
:param dialect: The dialect to which this will be encoded to
:return: The value encoded in that dialect
"""
if value is None:
return value
elif dialect.name == 'postgresql':
return str(value)
else:
return str(value).replace('-', '')
def validate(table=None, **kwargs):
"""
Validate that the passed arguments are sufficient for connecting to the
database.
:returns: True if the arguments should allow connecting to the db.
"""
if not table or not kwargs.get('uri'):
return False
# We Could validate other database parameters,too
return True
# Make a list of the dialects this module supports. There is no default
# dialect.
def upgrade():
"""Upgrade database."""
# The following is a ridiculous hack to force table recreation for sqlite to
# enable the use of a default timestamp.
recreate = 'auto'
migrate_context = context.get_context()
sqlite_dialect_class = None
if getattr(sa.dialects, 'sqlite', False):
sqlite_dialect_class = (sa.dialects.sqlite.pysqlite
.sqliteDialect_pysqlite)
if migrate_context.dialect.__class__ == sqlite_dialect_class:
recreate = 'always'
with op.batch_alter_table('games', recreate=recreate) as batch_op:
batch_op.add_column(sa.Column('played_at', sa.DateTime(),
nullable=False, server_default=sa.func.Now()))
def upgrade():
"""Upgrade database."""
# The following is a ridiculous hack to force table recreation for sqlite to
# enable the use of a default timestamp.
recreate = 'auto'
migrate_context = context.get_context()
sqlite_dialect_class = None
if getattr(sa.dialects, recreate=recreate) as batch_op:
batch_op.add_column(sa.Column('last_modified_at', server_default=sa.func.Now()))
def upgrade():
"""Upgrade database."""
# The following is a ridiculous hack to force table recreation for sqlite to
# enable the use of a default timestamp.
recreate = 'auto'
migrate_context = context.get_context()
sqlite_dialect_class = None
if getattr(sa.dialects, recreate=recreate) as batch_op:
batch_op.add_column(sa.Column('created_at', server_default=sa.func.Now()))
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。
