Method _execute_ddl

lib/sqlalchemy/engine/base.py:1523–1574

Execute a schema.DDL object.

(
        self,
        ddl: ExecutableDDLElement,
        distilled_parameters: _CoreMultiExecuteParams,
        execution_options: CoreExecuteOptionsParameter,
    )

        return ret

    def _execute_ddl(
        self,
        ddl: ExecutableDDLElement,
        distilled_parameters: _CoreMultiExecuteParams,
        execution_options: CoreExecuteOptionsParameter,
    ) -> CursorResult[Unpack[TupleAny]]:
        """Execute a schema.DDL object."""

        exec_opts = ddl._execution_options.merge_with(
            self._execution_options, execution_options
        )

        event_multiparams: Optional[_CoreMultiExecuteParams]
        event_params: Optional[_CoreSingleExecuteParams]

        if self._has_events or self.engine._has_events:
            (
                ddl,
                distilled_parameters,
                event_multiparams,
                event_params,
            ) = self._invoke_before_exec_event(
                ddl, distilled_parameters, exec_opts
            )
        else:
            event_multiparams = event_params = None

        schema_translate_map = exec_opts.get("schema_translate_map", None)

        dialect = self.dialect

        compiled = ddl.compile(
            dialect=dialect, schema_translate_map=schema_translate_map
        )
        ret = self._execute_context(
            dialect,
            dialect.execution_ctx_cls._init_ddl,
            compiled,
            None,
            exec_opts,
            compiled,
        )
        if self._has_events or self.engine._has_events:
            self.dispatch.after_execute(
                self,
                ddl,
                event_multiparams,
                event_params,
                exec_opts,
                ret,
            )
        return ret

    def _invoke_before_exec_event(
        self,