Method _enum_query

lib/sqlalchemy/dialects/postgresql/base.py:5471–5511

(self, schema)

    @lru_cache()
    def _enum_query(self, schema):
        lbl_agg_sq = (
            select(
                pg_catalog.pg_enum.c.enumtypid,
                sql.func.array_agg(
                    aggregate_order_by(
                        # NOTE: cast since some postgresql derivatives may
                        # not support array_agg on the name type
                        pg_catalog.pg_enum.c.enumlabel.cast(TEXT),
                        pg_catalog.pg_enum.c.enumsortorder,
                    )
                ).label("labels"),
            )
            .group_by(pg_catalog.pg_enum.c.enumtypid)
            .subquery("lbl_agg")
        )

        query = (
            select(
                pg_catalog.pg_type.c.typname.label("name"),
                pg_catalog.pg_type_is_visible(pg_catalog.pg_type.c.oid).label(
                    "visible"
                ),
                pg_catalog.pg_namespace.c.nspname.label("schema"),
                lbl_agg_sq.c.labels.label("labels"),
            )
            .join(
                pg_catalog.pg_namespace,
                pg_catalog.pg_namespace.c.oid
                == pg_catalog.pg_type.c.typnamespace,
            )
            .outerjoin(
                lbl_agg_sq, pg_catalog.pg_type.c.oid == lbl_agg_sq.c.enumtypid
            )
            .where(pg_catalog.pg_type.c.typtype == "e")
            .order_by(
                pg_catalog.pg_namespace.c.nspname, pg_catalog.pg_type.c.typname
            )
        )

        return self._pg_type_filter_schema(query, schema)

    @reflection.cache
    def _load_enums(self, connection, schema=None, **kw):