Method visit_cte

lib/sqlalchemy/sql/compiler.py:4234–4448

(
        self,
        cte: CTE,
        asfrom: bool = False,
        ashint: bool = False,
        fromhints: Optional[_FromHintsType] = None,
        visiting_cte: Optional[CTE] = None,
        from_linter: Optional[FromLinter] = None,
        cte_opts: selectable._CTEOpts = selectable._CTEOpts(False),
        **kwargs: Any,
    )

            cte._compiler_dispatch(self, cte_opts=opt, **local_kw)

    def visit_cte(
        self,
        cte: CTE,
        asfrom: bool = False,
        ashint: bool = False,
        fromhints: Optional[_FromHintsType] = None,
        visiting_cte: Optional[CTE] = None,
        from_linter: Optional[FromLinter] = None,
        cte_opts: selectable._CTEOpts = selectable._CTEOpts(False),
        **kwargs: Any,
    ) -> Optional[str]:
        self_ctes = self._init_cte_state()
        assert self_ctes is self.ctes

        kwargs["visiting_cte"] = cte

        cte_name = cte.name

        if isinstance(cte_name, elements._truncated_label):
            cte_name = self._truncated_identifier("alias", cte_name)

        is_new_cte = True
        embedded_in_current_named_cte = False

        _reference_cte = cte._get_reference_cte()

        nesting = cte.nesting or cte_opts.nesting

        # check for CTE already encountered
        if _reference_cte in self.level_name_by_cte:
            cte_level, _, existing_cte_opts = self.level_name_by_cte[
                _reference_cte
            ]
            assert _ == cte_name

            cte_level_name = (cte_level, cte_name)
            existing_cte = self.ctes_by_level_name[cte_level_name]

            # check if we are receiving it here with a specific
            # "nest_here" location; if so, move it to this location

            if cte_opts.nesting:
                if existing_cte_opts.nesting:
                    raise exc.CompileError(
                        "CTE is stated as 'nest_here' in "
                        "more than one location"
                    )

                old_level_name = (cte_level, cte_name)
                cte_level = len(self.stack) if nesting else 1
                cte_level_name = new_level_name = (cte_level, cte_name)

                del self.ctes_by_level_name[old_level_name]
                self.ctes_by_level_name[new_level_name] = existing_cte
                self.level_name_by_cte[_reference_cte] = new_level_name + (
                    cte_opts,
                )