Method _domain_query

lib/sqlalchemy/dialects/postgresql/base.py:5386–5443

(self, schema)

    @lru_cache()
    def _domain_query(self, schema):
        con_sq = (
            select(
                pg_catalog.pg_constraint.c.contypid,
                sql.func.array_agg(
                    pg_catalog.pg_get_constraintdef(
                        pg_catalog.pg_constraint.c.oid, True
                    )
                ).label("condefs"),
                sql.func.array_agg(
                    # NOTE: cast since some postgresql derivatives may
                    # not support array_agg on the name type
                    pg_catalog.pg_constraint.c.conname.cast(TEXT)
                ).label("connames"),
            )
            # The domain this constraint is on; zero if not a domain constraint
            .where(pg_catalog.pg_constraint.c.contypid != 0)
            .group_by(pg_catalog.pg_constraint.c.contypid)
            .subquery("domain_constraints")
        )

        query = (
            select(
                pg_catalog.pg_type.c.typname.label("name"),
                pg_catalog.format_type(
                    pg_catalog.pg_type.c.typbasetype,
                    pg_catalog.pg_type.c.typtypmod,
                ).label("attype"),
                (~pg_catalog.pg_type.c.typnotnull).label("nullable"),
                pg_catalog.pg_type.c.typdefault.label("default"),
                pg_catalog.pg_type_is_visible(pg_catalog.pg_type.c.oid).label(
                    "visible"
                ),
                pg_catalog.pg_namespace.c.nspname.label("schema"),
                con_sq.c.condefs,
                con_sq.c.connames,
                pg_catalog.pg_collation.c.collname,
            )
            .join(
                pg_catalog.pg_namespace,
                pg_catalog.pg_namespace.c.oid
                == pg_catalog.pg_type.c.typnamespace,
            )
            .outerjoin(
                pg_catalog.pg_collation,
                pg_catalog.pg_type.c.typcollation
                == pg_catalog.pg_collation.c.oid,
            )
            .outerjoin(
                con_sq,
                pg_catalog.pg_type.c.oid == con_sq.c.contypid,
            )
            .where(pg_catalog.pg_type.c.typtype == "d")
            .order_by(
                pg_catalog.pg_namespace.c.nspname, pg_catalog.pg_type.c.typname
            )
        )
        return self._pg_type_filter_schema(query, schema)