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

Code before:
            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

    def set_password(self, value: str | None):
        """Encrypt password and set in object attribute."""
        if value:
            fernet = get_fernet()
            self._password = fernet.encrypt(bytes(value, "utf-8")).decode()
            self.is_encrypted = fernet.is_encrypted

    @declared_attr
    def password(cls):
        """Password. The value is decrypted/encrypted when reading/setting the value."""

Review comment (refactor):
It seems the whole flow could be simplify as:

```python
can_flatten = True
# the for loop to check whether could we flatten 

try:
    query: str | None = urlencode(self.extra_dejson)
except TypeError:
    query = None

# we just need to add can_flatten condition check
if can_flatten and query and self.extra_dejson == dict(parse_qsl(query, keep_blank_values=True)):
    uri += ("?" if self.schema else "/?") + query
else:
# fallback to nested EXTRA_KEY
    uri += ("?" if self.schema else "/?") + urlencode({self.EXTRA_KEY: self.extra})
```

which also minimized the change.

Code after:
                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):
        """Encrypt password and set in object attribute."""
        if value:
            fernet = get_fernet()
            self._password = fernet.encrypt(bytes(value, "utf-8")).decode()
            self.is_encrypted = fernet.is_encrypted

    @declared_attr
    def password(cls):
        """Password. The value is decrypted/encrypted when reading/setting the value."""
        return synonym("_password", descriptor=property(cls.get_password, cls.set_password))