Python 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, 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.UUID())
        else:
            return dialect.type_descriptor(CHAR(32))

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
        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, '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('last_modified_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, '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('created_at', sa.DateTime(),
                            nullable=False, server_default=sa.func.now()))

def process_result_value(
            self, value: Optional[str], dialect: dialects
    ) -> Optional[dict]:
        """

        :param value: The value to process from the SQL query
        :param dialect: The dialect to use for the processing
        :return: The value as a UUID
        """
        if value is None:
            return value
        else:
            return json.loads(value)