Method has_type

lib/sqlalchemy/dialects/postgresql/base.py:3847–3866

(self, connection, type_name, schema=None, **kw)

    @reflection.cache
    def has_type(self, connection, type_name, schema=None, **kw):
        query = (
            select(pg_catalog.pg_type.c.typname)
            .join(
                pg_catalog.pg_namespace,
                pg_catalog.pg_namespace.c.oid
                == pg_catalog.pg_type.c.typnamespace,
            )
            .where(pg_catalog.pg_type.c.typname == type_name)
        )
        if schema is None:
            query = query.where(
                pg_catalog.pg_type_is_visible(pg_catalog.pg_type.c.oid),
                # ignore pg_catalog schema
                pg_catalog.pg_namespace.c.nspname != "pg_catalog",
            )
        elif schema != "*":
            query = query.where(pg_catalog.pg_namespace.c.nspname == schema)

        return bool(connection.scalar(query))

    def _get_server_version_info(self, connection):
        v = connection.exec_driver_sql("select pg_catalog.version()").scalar()