def get_settings_to_upload()

in core/maxframe_client/session/odps.py [0:0]

• 45 lines of code
• 18 McCabe index (conditional complexity)

    def get_settings_to_upload(self) -> Dict[str, Any]:
        sql_settings = (odps_options.sql.settings or {}).copy()
        sql_settings.update(options.sql.settings or {})
        quota_name = options.session.quota_name or getattr(
            odps_options, "quota_name", None
        )
        quota_settings = {
            sql_settings.get("odps.task.wlm.quota", None),
            options.spe.task.settings.get("odps.task.wlm.quota", None),
            options.pythonpack.task.settings.get("odps.task.wlm.quota", None),
            quota_name,
        }.difference([None])
        if len(quota_settings) >= 2:
            raise ValueError(
                "Quota settings are conflicting: %s" % ", ".join(sorted(quota_settings))
            )
        elif len(quota_settings) == 1:
            quota_name = quota_settings.pop()
        lifecycle = options.session.table_lifecycle or odps_options.lifecycle
        temp_lifecycle = (
            options.session.temp_table_lifecycle or odps_options.temp_lifecycle
        )

        enable_schema = options.session.enable_schema
        default_schema = options.session.default_schema
        if hasattr(self, "_odps_entry"):
            default_schema = default_schema or self._odps_entry.schema

        # use flags in sql settings
        if sql_settings.get("odps.default.schema"):
            default_schema = sql_settings["odps.default.schema"]
        if str_to_bool(
            sql_settings.get("odps.namespace.schema") or "false"
        ) or str_to_bool(
            sql_settings.get("odps.sql.allow.namespace.schema") or "false"
        ):
            enable_schema = True

        mf_settings = dict(options.to_dict(remote_only=True).items())
        mf_settings["sql.settings"] = sql_settings
        mf_settings["session.table_lifecycle"] = lifecycle
        mf_settings["session.temp_table_lifecycle"] = temp_lifecycle
        mf_settings["session.quota_name"] = quota_name
        if enable_schema is not None:
            mf_settings["session.enable_schema"] = enable_schema
        if options.session.enable_high_availability is None:
            mf_settings["session.enable_high_availability"] = not in_ipython_frontend()
        mf_settings["session.default_schema"] = default_schema or "default"
        return mf_settings