Repository: apache/airflow
File: airflow-core/src/airflow/models/connection.py (Python)

Code before:

            uri += authority_block

        host_block = ""
        if host_to_use:
            host_block += quote(host_to_use, safe="")

        if self.port:
            if host_block == "" and authority_block == "":
                host_block += f"@:{self.port}"
            else:
                host_block += f":{self.port}"

        if self.schema:
            host_block += f"/{quote(self.schema, safe='')}"

        uri += host_block

        if self.extra:
            try:
                extra_dict = self.extra_dejson
                can_flatten = True
                for _, value in extra_dict.items():
                    if not isinstance(value, str):
                        can_flatten = False
                        break

                if can_flatten:
                    query = urlencode(extra_dict)
                    if extra_dict == dict(parse_qsl(query, keep_blank_values=True)):
                        uri += ("?" if self.schema else "/?") + query
                    else:
                        uri += ("?" if self.schema else "/?") + urlencode({self.EXTRA_KEY: self.extra})
                else:
                    uri += ("?" if self.schema else "/?") + urlencode({self.EXTRA_KEY: self.extra})
            except (TypeError, AttributeError):
                uri += ("?" if self.schema else "/?") + urlencode({self.EXTRA_KEY: self.extra})
        return uri

    def get_password(self) -> str | None:
        """Return encrypted password."""
        if self._password and self.is_encrypted:
            fernet = get_fernet()
            if not fernet.is_encrypted:
                raise AirflowException(
                    f"Can't decrypt encrypted password for login={self.login}  "
                    f"FERNET_KEY configuration is missing"
                )
            return fernet.decrypt(bytes(self._password, "utf-8")).decode()
        return self._password


Review comment (suggestion):
How about moving the `can_flatten` logic before the `try` block, as it _should_ not raise any `TypeError, AttributeError` exception.

Code after:
            authority_block += "@"

            uri += authority_block

        host_block = ""
        if host_to_use:
            host_block += quote(host_to_use, safe="")

        if self.port:
            if host_block == "" and authority_block == "":
                host_block += f"@:{self.port}"
            else:
                host_block += f":{self.port}"

        if self.schema:
            host_block += f"/{quote(self.schema, safe='')}"

        uri += host_block

        if self.extra:
            extra_dict = self.extra_dejson
            can_flatten = True
            for value in extra_dict.values():
                if not isinstance(value, str):
                    can_flatten = False
                    break

            try:
                query: str | None = urlencode(extra_dict)
            except TypeError:
                query = None

            if can_flatten and query and extra_dict == dict(parse_qsl(query, keep_blank_values=True)):
                uri += ("?" if self.schema else "/?") + query
            else:
                uri += ("?" if self.schema else "/?") + urlencode({self.EXTRA_KEY: self.extra})
        return uri

    def get_password(self) -> str | None:
        """Return encrypted password."""
        if self._password and self.is_encrypted:
            fernet = get_fernet()
            if not fernet.is_encrypted:
                raise AirflowException(
                    f"Can't decrypt encrypted password for login={self.login}  "
                    f"FERNET_KEY configuration is missing"
                )
            return fernet.decrypt(bytes(self._password, "utf-8")).decode()
        return self._password

    def set_password(self, value: str | None):