Repository: apache/airflow
File: providers/vertica/src/airflow/providers/vertica/hooks/vertica.py (Python)

Code before:

        if "log_level" in conn_extra:
            import logging

            log_lvl = conn_extra["log_level"]
            conn_config["log_path"] = None
            if isinstance(log_lvl, str):
                log_lvl = log_lvl.lower()
                if log_lvl == "critical":
                    conn_config["log_level"] = logging.CRITICAL
                elif log_lvl == "error":
                    conn_config["log_level"] = logging.ERROR
                elif log_lvl == "warning":
                    conn_config["log_level"] = logging.WARNING
                elif log_lvl == "info":
                    conn_config["log_level"] = logging.INFO
                elif log_lvl == "debug":
                    conn_config["log_level"] = logging.DEBUG
                elif log_lvl == "notset":
                    conn_config["log_level"] = logging.NOTSET
            else:
                conn_config["log_level"] = int(conn_extra["log_level"])

        conn = connect(**conn_config)
        return conn

    @property
    def sqlalchemy_url(self) -> URL:
        """Return a SQLAlchemy URL object with properly formatted query parameters."""
        if URL is None:
            raise AirflowOptionalProviderFeatureException(
                "The 'sqlalchemy' library is required to use 'sqlalchemy_url'."
                "Please install it with: pip install 'apache-airflow-providers-vertica[sqlalchemy]'"
            )
        conn = self.get_connection(self.get_conn_id())
        extra = conn.extra_dejson or {}

        # Normalize query dictionary
        query = {
            k: ([str(x) for x in v] if isinstance(v, (list, tuple)) else str(v))
            for k, v in extra.items()
            if v is not None
        }

        return URL.create(
            drivername="vertica-python",
            username=conn.login,
            password=conn.password or "",
            host=conn.host or "localhost",
            port=conn.port or 5433,
            database=conn.schema,

Review comment (bug):
I think it will fail `mypy` tests now that we moved to sqlalchemy 2. We need to add typechecking exclusion likely.

Code after:
            import logging

            log_lvl = conn_extra["log_level"]
            conn_config["log_path"] = None
            if isinstance(log_lvl, str):
                log_lvl = log_lvl.lower()
                if log_lvl == "critical":
                    conn_config["log_level"] = logging.CRITICAL
                elif log_lvl == "error":
                    conn_config["log_level"] = logging.ERROR
                elif log_lvl == "warning":
                    conn_config["log_level"] = logging.WARNING
                elif log_lvl == "info":
                    conn_config["log_level"] = logging.INFO
                elif log_lvl == "debug":
                    conn_config["log_level"] = logging.DEBUG
                elif log_lvl == "notset":
                    conn_config["log_level"] = logging.NOTSET
            else:
                conn_config["log_level"] = int(conn_extra["log_level"])

        conn = connect(**conn_config)
        return conn

    @property
    def sqlalchemy_url(self) -> URL:
        """Return a SQLAlchemy URL object with properly formatted query parameters."""
        try:
            from sqlalchemy.engine import URL
        except (ImportError, ModuleNotFoundError) as err:
            raise AirflowOptionalProviderFeatureException(
                "The 'sqlalchemy' library is required to use this feature. "
                "Please install it with: pip install 'apache-airflow-providers-vertica[sqlalchemy]'"
            ) from err
        conn = self.get_connection(self.get_conn_id())
        extra = conn.extra_dejson or {}

        # Normalize query dictionary
        query = {
            k: ([str(x) for x in v] if isinstance(v, (list, tuple)) else str(v))
            for k, v in extra.items()
            if v is not None
        }

        return URL.create(
            drivername="vertica-python",
            username=conn.login,
            password=conn.password or "",
            host=conn.host or "localhost",
            port=conn.port or 5433,
            database=conn.schema,