Make conn_type optional in task SDK Connection datamodel

[amoghrajesh]

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

closes: #58529

Why?

When workers retrieve connections from secrets backends like AWS Secrets Manager that don't include conn_type, the Connection.from_json() method fails during deserialization with error:

TypeError: Connection.__init__() missing 1 required keyword-only argument: 'conn_type'

Simple test to show an example for this:

(apache-airflow) ➜  airflow git:(conn-type-optional-sdk) ✗ cat > /tmp/test_connections.json << 'EOF'
{
  "my_db": {
    "host": "mydb.us-east-1.rds.amazonaws.com",
    "port": 5432,
    "login": "admin",
    "password": "secret123",
    "schema": "production"
  }
}
EOF

Before:

>>> import json
>>> json_from_aws = json.dumps({
...     "host": "mydb.us-east-1.rds.amazonaws.com",
...     "port": 5432,
...     "login": "admin",
...     "password": "secret123",
...     "schema": "production"
... })
>>> 
>>> from airflow.sdk import Connection
>>> conn = Connection.from_json(json_from_aws, conn_id="my_db")
Traceback (most recent call last):
  File "<python-input-10>", line 1, in <module>
    conn = Connection.from_json(json_from_aws, conn_id="my_db")
  File "/Users/amoghdesai/Documents/OSS/repos/airflow/task-sdk/src/airflow/sdk/definitions/connection.py", line 307, in from_json
    return cls(conn_id=conn_id, **kwargs)
TypeError: Connection.__init__() missing 1 required positional argument: 'conn_type'

After:

>>> import json
>>> from airflow.sdk import Connection
>>> 
>>> json_from_aws = json.dumps({
...     "host": "mydb.us-east-1.rds.amazonaws.com",
...     "port": 5432,
...     "login": "admin",
...     "password": "secret123",
...     "schema": "production"
... })
>>> conn = Connection.from_json(json_from_aws, conn_id="my_db")
>>> conn
Connection(conn_id='my_db', conn_type=None, description=None, host='mydb.us-east-1.rds.amazonaws.com', schema='production', login='admin', password='secret123', port=5432, extra=None)

What

Making conn_type optional in task SDK Connection datamodel to restore backward compatibility with Airflow 2 for connections stored in secrets backends without conn_type.

For example: connections in AWS Secrets Manager (and other external secrets backends) fail to load on workers when conn_type field is missing and that breaks compat with Airflow 3 for certain providers.

This will later block users migrating from Airflow 2 to Airflow 3 who have hundreds/thousands of secrets stored without conn_type (which worked fine in Airflow 2).

Solution

Make conn_type optional in task SDK Connection class. Let me explain why this is safe.

1. Secrets backends run only on workers - connections without conn_type exist only in worker context, never sent via Execution API
2. The responses that are sent by the API server always has conn_type - connections from database always have conn_type (NOT NULL column), so execution API ConnectionResponse doesn't need changes at all (cadwyn migration etc)

For testing, I have added one scenario that would break earlier but doesn't anymore.

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[amoghrajesh]

Actually this is a bug fix, which we can get into 3.1.8 as well