Consolidate Pydantic to ORM update in connections API

Author: jscheffl

airflow-core/src/airflow/api_fastapi/core_api/routes/public/connections.py

@@ -44,7 +44,10 @@
 )
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
 from airflow.api_fastapi.core_api.security import requires_access_connection
-from airflow.api_fastapi.core_api.services.public.connections import BulkConnectionService
+from airflow.api_fastapi.core_api.services.public.connections import (
+    BulkConnectionService,
+    update_orm_from_pydantic,
+)
 from airflow.api_fastapi.logging.decorators import action_logging
 from airflow.configuration import conf
 from airflow.models import Connection
@@ -200,23 +203,7 @@ def patch_connection(
         raise RequestValidationError(errors=e.errors())
 
     # Not all fields match and some need setters, therefore copy manually
-    if not update_mask or "conn_type" in update_mask:
-        connection.conn_type = patch_body.conn_type
-    if not update_mask or "description" in update_mask:
-        connection.description = patch_body.description
-    if not update_mask or "host" in update_mask:
-        connection.host = patch_body.host
-    if not update_mask or "schema" in update_mask:
-        connection.schema = patch_body.schema_
-    if not update_mask or "login" in update_mask:
-        connection.login = patch_body.login
-    if not update_mask or "password" in update_mask:
-        connection.set_password(patch_body.password)
-    if not update_mask or "port" in update_mask:
-        connection.port = patch_body.port
-    if not update_mask or "extra" in update_mask:
-        connection.set_extra(patch_body.extra)
-
+    update_orm_from_pydantic(connection, patch_body, update_mask)
     return connection
 
 

airflow-core/src/airflow/api_fastapi/core_api/services/public/connections.py

@@ -34,6 +34,29 @@
 from airflow.models.connection import Connection
 
 
+def update_orm_from_pydantic(
+    orm_conn: Connection, pydantic_conn: ConnectionBody, update_mask: list[str] | None = None
+):
+    """Update ORM object from Pydantic object."""
+    # Not all fields match and some need setters, therefore copy manually
+    if not update_mask or "conn_type" in update_mask:
+        orm_conn.conn_type = pydantic_conn.conn_type
+    if not update_mask or "description" in update_mask:
+        orm_conn.description = pydantic_conn.description
+    if not update_mask or "host" in update_mask:
+        orm_conn.host = pydantic_conn.host
+    if not update_mask or "schema" in update_mask:
+        orm_conn.schema = pydantic_conn.schema_
+    if not update_mask or "login" in update_mask:
+        orm_conn.login = pydantic_conn.login
+    if not update_mask or "password" in update_mask:
+        orm_conn.set_password(pydantic_conn.password)
+    if not update_mask or "port" in update_mask:
+        orm_conn.port = pydantic_conn.port
+    if not update_mask or "extra" in update_mask:
+        orm_conn.set_extra(pydantic_conn.extra)
+
+
 class BulkConnectionService(BulkService[ConnectionBody]):
     """Service for handling bulk operations on connections."""
 
@@ -118,14 +141,7 @@ def handle_bulk_update(
                     ConnectionBody(**connection.model_dump())
 
                     # Not all fields match and some need setters, therefore copy manually
-                    old_connection.conn_type = connection.conn_type
-                    old_connection.description = connection.description
-                    old_connection.host = connection.host
-                    old_connection.schema = connection.schema_
-                    old_connection.login = connection.login
-                    old_connection.set_password(connection.password)
-                    old_connection.port = connection.port
-                    old_connection.set_extra(connection.extra)
+                    update_orm_from_pydantic(old_connection, connection)
                     results.success.append(connection.connection_id)
 
         except HTTPException as e: