jason810496 commented on code in PR #53189:
URL: https://github.com/apache/airflow/pull/53189#discussion_r2202354286
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py:
##########
@@ -272,3 +293,456 @@ def get_hitl_details(
hitl_details=hitl_details,
total_entries=total_entries,
)
+
+
+@hitl_router.post(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}",
+ status_code=status.HTTP_201_CREATED,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+ dependencies=[Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.TASK_INSTANCE))],
+)
+def create_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ user: GetUserDep,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Create a shared link for a Human-in-the-loop task.
+
+ This endpoint generates a secure, time-limited shared link that allows
external users
+ to interact with HITL tasks without requiring full Airflow authentication.
The link
+ can be configured for either direct action execution or UI redirection.
+
+ :param dag_id: The DAG identifier
+ :param dag_run_id: The DAG run identifier
+ :param task_id: The task identifier
+ :param update_hitl_detail_payload: Payload containing link configuration
and initial response data
+ :param user: The authenticated user creating the shared link
+ :param session: Database session for data persistence
+
+ :raises HTTPException: 403 if HITL shared links are not enabled
+ :raises HTTPException: 404 if the task instance or HITL detail does not
exist
+ :raises HTTPException: 400 if link generation fails due to invalid
parameters
+
+ :return: HITLDetailResponse containing the generated link URL and metadata
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ task_instance = _get_task_instance(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ session=session,
+ map_index=None,
+ )
+
+ ti_id_str = str(task_instance.id)
+ hitl_detail_model =
session.scalar(select(HITLDetailModel).where(HITLDetailModel.ti_id ==
ti_id_str))
+ if not hitl_detail_model:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Human-in-the-loop detail does not exist for Task Instance with
id {ti_id_str}",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.generate_link(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ map_index=None,
+ link_type=update_hitl_detail_payload.link_type,
+ action=update_hitl_detail_payload.action,
+ expires_in_hours=update_hitl_detail_payload.expires_in_hours,
+ )
+
+ response = HITLDetailResponse(
+ user_id=user.get_id(),
+ response_at=timezone.utcnow(),
+ chosen_options=update_hitl_detail_payload.chosen_options,
+ params_input=update_hitl_detail_payload.params_input,
+ task_instance_id=link_data["task_instance_id"],
+ link_url=link_data["link_url"],
+ expires_at=link_data["expires_at"],
+ action=link_data["action"],
+ link_type=link_data["link_type"],
+ )
+
+ return response
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
Review Comment:
Will having a new `BaseErrorHandler` like what we do for
`DatabaseErrorHandlers` a good idea for handling `valueError` for all routes?
Or handling with existed try / except will be enough?
Any suggestion for this one ?
cc @pierrejeambrun
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/hitl.py:
##########
@@ -32,6 +32,17 @@ class UpdateHITLDetailPayload(BaseModel):
chosen_options: list[str]
params_input: Mapping = Field(default_factory=dict)
+ # Shared link fields
+ link_type: str = Field(
+ default="action",
+ description="Type of link to generate: 'action' for direct action or
'redirect' for UI interaction",
+ )
+ action: str | None = Field(
+ default=None,
+ description="Optional action to perform when link is accessed (e.g.,
'approve', 'reject'). Required for action links.",
+ )
+ expires_in_hours: int | None = Field(default=None, description="Optional
custom expiration time in hours")
+
Review Comment:
How about to have a subclass for common fields for both
`UpdateHITLDetailPayload` and `HITLDetail`.
##########
airflow-core/src/airflow/utils/hitl_shared_links.py:
##########
@@ -0,0 +1,220 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Utilities for Human-in-the-Loop (HITL) shared links."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+from datetime import datetime, timedelta
+from typing import Any
+from urllib.parse import urlencode
+
+import structlog
+
+from airflow.configuration import conf
+from airflow.utils import timezone
+
+log = structlog.get_logger(__name__)
+
+
+class HITLSharedLinkManager:
+ """Manager for HITL shared links with token generation and verification."""
+
+ def __init__(self):
+ self.secret_key = conf.get("api", "hitl_shared_link_secret_key",
fallback="")
+ self.default_expiration_hours = conf.getint("api",
"hitl_shared_link_expiration_hours", fallback=24)
+
+ def is_enabled(self) -> bool:
+ """Check if HITL shared links are enabled."""
+ return conf.getboolean("api", "hitl_enable_shared_links",
fallback=False)
+
+ def _generate_signature(self, payload: str) -> str:
+ """Generate HMAC signature for the payload."""
+ if not self.secret_key:
+ raise ValueError("HITL shared link secret key is not configured")
+
+ signature = hmac.new(
+ self.secret_key.encode("utf-8"), payload.encode("utf-8"),
hashlib.sha256
+ ).digest()
+ return base64.urlsafe_b64encode(signature).decode("utf-8")
+
+ def _verify_signature(self, payload: str, signature: str) -> bool:
+ """Verify HMAC signature for the payload."""
+ expected_signature = self._generate_signature(payload)
+ return hmac.compare_digest(expected_signature, signature)
+
+ def generate_link(
+ self,
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int | None = None,
+ link_type: str = "action",
+ action: str | None = None,
+ expires_in_hours: int | None = None,
+ base_url: str | None = None,
+ ) -> dict[str, Any]:
+ """
+ Generate a shared link for HITL task.
+
+ :param dag_id: DAG ID
+ :param dag_run_id: DAG run ID
+ :param task_id: Task ID
+ :param map_index: Map index for mapped tasks
+ :param link_type: Type of link ('action' or 'redirect')
+ :param action: Action to perform (for action links)
+ :param expires_in_hours: Custom expiration time in hours
+ :param base_url: Base URL for the link
+ """
+ if not self.is_enabled():
+ raise ValueError("HITL shared links are not enabled")
+
+ if link_type == "action" and not action:
+ raise ValueError("Action is required for action-type links")
+
+ expiration_hours = expires_in_hours or self.default_expiration_hours
+ expires_at = timezone.utcnow() + timedelta(hours=expiration_hours)
+
+ payload_data = {
+ "dag_id": dag_id,
+ "dag_run_id": dag_run_id,
+ "task_id": task_id,
+ "map_index": map_index,
+ "link_type": link_type,
+ "action": action,
+ "expires_at": expires_at.isoformat(),
+ }
Review Comment:
How about type the `payload_data` with something like
`HITLSharedPayload(TypedDict)` ?
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py:
##########
@@ -272,3 +293,456 @@ def get_hitl_details(
hitl_details=hitl_details,
total_entries=total_entries,
)
+
+
+@hitl_router.post(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}",
+ status_code=status.HTTP_201_CREATED,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+ dependencies=[Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.TASK_INSTANCE))],
+)
+def create_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ user: GetUserDep,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Create a shared link for a Human-in-the-loop task.
+
+ This endpoint generates a secure, time-limited shared link that allows
external users
+ to interact with HITL tasks without requiring full Airflow authentication.
The link
+ can be configured for either direct action execution or UI redirection.
+
+ :param dag_id: The DAG identifier
+ :param dag_run_id: The DAG run identifier
+ :param task_id: The task identifier
+ :param update_hitl_detail_payload: Payload containing link configuration
and initial response data
+ :param user: The authenticated user creating the shared link
+ :param session: Database session for data persistence
+
+ :raises HTTPException: 403 if HITL shared links are not enabled
+ :raises HTTPException: 404 if the task instance or HITL detail does not
exist
+ :raises HTTPException: 400 if link generation fails due to invalid
parameters
+
+ :return: HITLDetailResponse containing the generated link URL and metadata
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ task_instance = _get_task_instance(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ session=session,
+ map_index=None,
+ )
+
+ ti_id_str = str(task_instance.id)
+ hitl_detail_model =
session.scalar(select(HITLDetailModel).where(HITLDetailModel.ti_id ==
ti_id_str))
+ if not hitl_detail_model:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Human-in-the-loop detail does not exist for Task Instance with
id {ti_id_str}",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.generate_link(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ map_index=None,
+ link_type=update_hitl_detail_payload.link_type,
+ action=update_hitl_detail_payload.action,
+ expires_in_hours=update_hitl_detail_payload.expires_in_hours,
+ )
+
+ response = HITLDetailResponse(
+ user_id=user.get_id(),
+ response_at=timezone.utcnow(),
+ chosen_options=update_hitl_detail_payload.chosen_options,
+ params_input=update_hitl_detail_payload.params_input,
+ task_instance_id=link_data["task_instance_id"],
+ link_url=link_data["link_url"],
+ expires_at=link_data["expires_at"],
+ action=link_data["action"],
+ link_type=link_data["link_type"],
+ )
+
+ return response
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.post(
+
"/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}/{map_index}",
+ status_code=status.HTTP_201_CREATED,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+ dependencies=[Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.TASK_INSTANCE))],
+)
+def create_mapped_ti_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ user: GetUserDep,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Create a shared link for a mapped Human-in-the-loop task.
+
+ This endpoint generates a secure, time-limited shared link for mapped task
instances,
+ allowing external users to interact with specific mapped HITL tasks
without requiring
+ full Airflow authentication. The link can be configured for either direct
action
+ execution or UI redirection.
+
+ :param dag_id: The DAG identifier
+ :param dag_run_id: The DAG run identifier
+ :param task_id: The task identifier
+ :param map_index: The map index for the mapped task instance
+ :param update_hitl_detail_payload: Payload containing link configuration
and initial response data
+ :param user: The authenticated user creating the shared link
+ :param session: Database session for data persistence
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ task_instance = _get_task_instance(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ session=session,
+ map_index=map_index,
+ )
+
+ ti_id_str = str(task_instance.id)
+ hitl_detail_model =
session.scalar(select(HITLDetailModel).where(HITLDetailModel.ti_id ==
ti_id_str))
+ if not hitl_detail_model:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Human-in-the-loop detail does not exist for Task Instance with
id {ti_id_str}",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.generate_link(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ map_index=map_index,
+ link_type=update_hitl_detail_payload.link_type,
+ action=update_hitl_detail_payload.action,
+ expires_in_hours=update_hitl_detail_payload.expires_in_hours,
+ )
+
+ response = HITLDetailResponse(
+ user_id=user.get_id(),
+ response_at=timezone.utcnow(),
+ chosen_options=update_hitl_detail_payload.chosen_options,
+ params_input=update_hitl_detail_payload.params_input,
+ task_instance_id=link_data["task_instance_id"],
+ link_url=link_data["link_url"],
+ expires_at=link_data["expires_at"],
+ action=link_data["action"],
+ link_type=link_data["link_type"],
+ )
+
+ return response
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.get(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}",
+ status_code=status.HTTP_200_OK,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
Review Comment:
```suggestion
responses=create_openapi_http_exception_doc(
[
status.HTTP_400_BAD_REQUEST,
status.HTTP_403_FORBIDDEN,
status.HTTP_404_NOT_FOUND,
]
),
```
The 200 status code is default by FastAPI itself.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py:
##########
@@ -272,3 +293,456 @@ def get_hitl_details(
hitl_details=hitl_details,
total_entries=total_entries,
)
+
+
+@hitl_router.post(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}",
+ status_code=status.HTTP_201_CREATED,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
Review Comment:
```suggestion
[
status.HTTP_400_BAD_REQUEST,
status.HTTP_403_FORBIDDEN,
status.HTTP_404_NOT_FOUND,
]
```
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py:
##########
@@ -75,10 +85,21 @@ def _update_hitl_detail(
dag_run_id: str,
task_id: str,
update_hitl_detail_payload: UpdateHITLDetailPayload,
- user: GetUserDep,
+ user: GetUserDep | None,
session: SessionDep,
map_index: int | None = None,
) -> HITLDetailResponse:
+ """
Review Comment:
We should move these function
- `_get_task_instance`
- `_update_hitl_detail`
- `_get_hitl_detail`
under `airflow-core/src/airflow/api_fastapi/core_api/services/public`
##########
airflow-core/src/airflow/utils/hitl_shared_links.py:
##########
@@ -0,0 +1,220 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Utilities for Human-in-the-Loop (HITL) shared links."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+from datetime import datetime, timedelta
+from typing import Any
+from urllib.parse import urlencode
+
+import structlog
+
+from airflow.configuration import conf
+from airflow.utils import timezone
+
+log = structlog.get_logger(__name__)
+
+
+class HITLSharedLinkManager:
+ """Manager for HITL shared links with token generation and verification."""
+
+ def __init__(self):
+ self.secret_key = conf.get("api", "hitl_shared_link_secret_key",
fallback="")
+ self.default_expiration_hours = conf.getint("api",
"hitl_shared_link_expiration_hours", fallback=24)
Review Comment:
The default value is set to `24` in `config.yaml`, do we still need the
`fallback=24` here ?
##########
airflow-core/src/airflow/utils/hitl_shared_links.py:
##########
@@ -0,0 +1,220 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Utilities for Human-in-the-Loop (HITL) shared links."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+from datetime import datetime, timedelta
+from typing import Any
+from urllib.parse import urlencode
+
+import structlog
+
+from airflow.configuration import conf
+from airflow.utils import timezone
+
+log = structlog.get_logger(__name__)
+
+
+class HITLSharedLinkManager:
+ """Manager for HITL shared links with token generation and verification."""
+
+ def __init__(self):
+ self.secret_key = conf.get("api", "hitl_shared_link_secret_key",
fallback="")
+ self.default_expiration_hours = conf.getint("api",
"hitl_shared_link_expiration_hours", fallback=24)
+
+ def is_enabled(self) -> bool:
+ """Check if HITL shared links are enabled."""
+ return conf.getboolean("api", "hitl_enable_shared_links",
fallback=False)
+
+ def _generate_signature(self, payload: str) -> str:
+ """Generate HMAC signature for the payload."""
+ if not self.secret_key:
+ raise ValueError("HITL shared link secret key is not configured")
+
+ signature = hmac.new(
+ self.secret_key.encode("utf-8"), payload.encode("utf-8"),
hashlib.sha256
+ ).digest()
+ return base64.urlsafe_b64encode(signature).decode("utf-8")
+
+ def _verify_signature(self, payload: str, signature: str) -> bool:
+ """Verify HMAC signature for the payload."""
+ expected_signature = self._generate_signature(payload)
+ return hmac.compare_digest(expected_signature, signature)
+
+ def generate_link(
+ self,
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int | None = None,
+ link_type: str = "action",
+ action: str | None = None,
+ expires_in_hours: int | None = None,
+ base_url: str | None = None,
+ ) -> dict[str, Any]:
+ """
+ Generate a shared link for HITL task.
+
+ :param dag_id: DAG ID
+ :param dag_run_id: DAG run ID
+ :param task_id: Task ID
+ :param map_index: Map index for mapped tasks
+ :param link_type: Type of link ('action' or 'redirect')
+ :param action: Action to perform (for action links)
+ :param expires_in_hours: Custom expiration time in hours
+ :param base_url: Base URL for the link
+ """
+ if not self.is_enabled():
+ raise ValueError("HITL shared links are not enabled")
+
+ if link_type == "action" and not action:
+ raise ValueError("Action is required for action-type links")
+
+ expiration_hours = expires_in_hours or self.default_expiration_hours
+ expires_at = timezone.utcnow() + timedelta(hours=expiration_hours)
+
+ payload_data = {
+ "dag_id": dag_id,
+ "dag_run_id": dag_run_id,
+ "task_id": task_id,
+ "map_index": map_index,
+ "link_type": link_type,
+ "action": action,
+ "expires_at": expires_at.isoformat(),
+ }
+
+ payload_str = json.dumps(payload_data, sort_keys=True)
+ signature = self._generate_signature(payload_str)
+
+ encoded_payload =
base64.urlsafe_b64encode(payload_str.encode("utf-8")).decode("utf-8")
+
+ if base_url is None:
+ base_url = conf.get("api", "base_url",
fallback="http://localhost:8080";)
+
+ if map_index is not None:
+ url_path =
f"/api/v2/hitl-details/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}/{map_index}"
+ else:
+ url_path =
f"/api/v2/hitl-details/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}"
+
+ query_params = {
+ "payload": encoded_payload,
+ "signature": signature,
+ }
+
+ link_url =
f"{base_url.rstrip('/')}{url_path}?{urlencode(query_params)}"
+
+ return {
+ "task_instance_id": f"{dag_id}.{dag_run_id}.{task_id}.{map_index
or -1}",
+ "link_url": link_url,
+ "expires_at": expires_at,
+ "action": action,
+ "link_type": link_type,
+ }
Review Comment:
And maybe have the strict type for the return dict as well IMHO.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py:
##########
@@ -272,3 +293,456 @@ def get_hitl_details(
hitl_details=hitl_details,
total_entries=total_entries,
)
+
+
+@hitl_router.post(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}",
+ status_code=status.HTTP_201_CREATED,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+ dependencies=[Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.TASK_INSTANCE))],
+)
+def create_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ user: GetUserDep,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Create a shared link for a Human-in-the-loop task.
+
+ This endpoint generates a secure, time-limited shared link that allows
external users
+ to interact with HITL tasks without requiring full Airflow authentication.
The link
+ can be configured for either direct action execution or UI redirection.
+
+ :param dag_id: The DAG identifier
+ :param dag_run_id: The DAG run identifier
+ :param task_id: The task identifier
+ :param update_hitl_detail_payload: Payload containing link configuration
and initial response data
+ :param user: The authenticated user creating the shared link
+ :param session: Database session for data persistence
+
+ :raises HTTPException: 403 if HITL shared links are not enabled
+ :raises HTTPException: 404 if the task instance or HITL detail does not
exist
+ :raises HTTPException: 400 if link generation fails due to invalid
parameters
+
+ :return: HITLDetailResponse containing the generated link URL and metadata
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ task_instance = _get_task_instance(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ session=session,
+ map_index=None,
+ )
+
+ ti_id_str = str(task_instance.id)
+ hitl_detail_model =
session.scalar(select(HITLDetailModel).where(HITLDetailModel.ti_id ==
ti_id_str))
+ if not hitl_detail_model:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Human-in-the-loop detail does not exist for Task Instance with
id {ti_id_str}",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.generate_link(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ map_index=None,
+ link_type=update_hitl_detail_payload.link_type,
+ action=update_hitl_detail_payload.action,
+ expires_in_hours=update_hitl_detail_payload.expires_in_hours,
+ )
+
+ response = HITLDetailResponse(
+ user_id=user.get_id(),
+ response_at=timezone.utcnow(),
+ chosen_options=update_hitl_detail_payload.chosen_options,
+ params_input=update_hitl_detail_payload.params_input,
+ task_instance_id=link_data["task_instance_id"],
+ link_url=link_data["link_url"],
+ expires_at=link_data["expires_at"],
+ action=link_data["action"],
+ link_type=link_data["link_type"],
+ )
+
+ return response
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.post(
+
"/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}/{map_index}",
+ status_code=status.HTTP_201_CREATED,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
Review Comment:
```suggestion
status.HTTP_400_BAD_REQUEST,
status.HTTP_403_FORBIDDEN,
status.HTTP_404_NOT_FOUND,
```
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -1493,6 +1493,32 @@ api:
type: boolean
example: ~
default: "False"
+ hitl_enable_shared_links:
+ description: |
+ Enable Human-in-the-Loop (HITL) shared links functionality. When
enabled, users can generate
+ shareable links for HITL tasks that can be used to perform actions or
redirect to the UI.
+ This feature must be explicitly enabled for security reasons.
+ version_added: 3.1.0
+ type: boolean
+ example: ~
+ default: "False"
+ hitl_shared_link_secret_key:
Review Comment:
+1 Agree with reusing `[api] secret_key`
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py:
##########
@@ -272,3 +293,456 @@ def get_hitl_details(
hitl_details=hitl_details,
total_entries=total_entries,
)
+
+
+@hitl_router.post(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}",
+ status_code=status.HTTP_201_CREATED,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+ dependencies=[Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.TASK_INSTANCE))],
+)
+def create_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ user: GetUserDep,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Create a shared link for a Human-in-the-loop task.
+
+ This endpoint generates a secure, time-limited shared link that allows
external users
+ to interact with HITL tasks without requiring full Airflow authentication.
The link
+ can be configured for either direct action execution or UI redirection.
+
+ :param dag_id: The DAG identifier
+ :param dag_run_id: The DAG run identifier
+ :param task_id: The task identifier
+ :param update_hitl_detail_payload: Payload containing link configuration
and initial response data
+ :param user: The authenticated user creating the shared link
+ :param session: Database session for data persistence
+
+ :raises HTTPException: 403 if HITL shared links are not enabled
+ :raises HTTPException: 404 if the task instance or HITL detail does not
exist
+ :raises HTTPException: 400 if link generation fails due to invalid
parameters
+
+ :return: HITLDetailResponse containing the generated link URL and metadata
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ task_instance = _get_task_instance(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ session=session,
+ map_index=None,
+ )
+
+ ti_id_str = str(task_instance.id)
+ hitl_detail_model =
session.scalar(select(HITLDetailModel).where(HITLDetailModel.ti_id ==
ti_id_str))
+ if not hitl_detail_model:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Human-in-the-loop detail does not exist for Task Instance with
id {ti_id_str}",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.generate_link(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ map_index=None,
+ link_type=update_hitl_detail_payload.link_type,
+ action=update_hitl_detail_payload.action,
+ expires_in_hours=update_hitl_detail_payload.expires_in_hours,
+ )
+
+ response = HITLDetailResponse(
+ user_id=user.get_id(),
+ response_at=timezone.utcnow(),
+ chosen_options=update_hitl_detail_payload.chosen_options,
+ params_input=update_hitl_detail_payload.params_input,
+ task_instance_id=link_data["task_instance_id"],
+ link_url=link_data["link_url"],
+ expires_at=link_data["expires_at"],
+ action=link_data["action"],
+ link_type=link_data["link_type"],
+ )
+
+ return response
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.post(
+
"/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}/{map_index}",
+ status_code=status.HTTP_201_CREATED,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+ dependencies=[Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.TASK_INSTANCE))],
+)
+def create_mapped_ti_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ user: GetUserDep,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Create a shared link for a mapped Human-in-the-loop task.
+
+ This endpoint generates a secure, time-limited shared link for mapped task
instances,
+ allowing external users to interact with specific mapped HITL tasks
without requiring
+ full Airflow authentication. The link can be configured for either direct
action
+ execution or UI redirection.
+
+ :param dag_id: The DAG identifier
+ :param dag_run_id: The DAG run identifier
+ :param task_id: The task identifier
+ :param map_index: The map index for the mapped task instance
+ :param update_hitl_detail_payload: Payload containing link configuration
and initial response data
+ :param user: The authenticated user creating the shared link
+ :param session: Database session for data persistence
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ task_instance = _get_task_instance(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ session=session,
+ map_index=map_index,
+ )
+
+ ti_id_str = str(task_instance.id)
+ hitl_detail_model =
session.scalar(select(HITLDetailModel).where(HITLDetailModel.ti_id ==
ti_id_str))
+ if not hitl_detail_model:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Human-in-the-loop detail does not exist for Task Instance with
id {ti_id_str}",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.generate_link(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ map_index=map_index,
+ link_type=update_hitl_detail_payload.link_type,
+ action=update_hitl_detail_payload.action,
+ expires_in_hours=update_hitl_detail_payload.expires_in_hours,
+ )
+
+ response = HITLDetailResponse(
+ user_id=user.get_id(),
+ response_at=timezone.utcnow(),
+ chosen_options=update_hitl_detail_payload.chosen_options,
+ params_input=update_hitl_detail_payload.params_input,
+ task_instance_id=link_data["task_instance_id"],
+ link_url=link_data["link_url"],
+ expires_at=link_data["expires_at"],
+ action=link_data["action"],
+ link_type=link_data["link_type"],
+ )
+
+ return response
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.get(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}",
+ status_code=status.HTTP_200_OK,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+)
+def get_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ payload: str,
+ signature: str,
+ session: SessionDep,
+) -> HITLDetail:
+ """
+ Get HITL details via shared link (for redirect links).
+
+ This endpoint allows external users to access HITL task details through a
secure
+ shared link. The link must be a redirect-type link, which provides
read-only access
+ to the HITL task information for UI rendering or decision-making purposes.
+
+ :param dag_id: The DAG identifier (from URL path)
+ :param dag_run_id: The DAG run identifier (from URL path)
+ :param task_id: The task identifier (from URL path)
+ :param payload: Base64-encoded payload containing link metadata and
expiration
+ :param signature: HMAC signature for payload verification
+ :param session: Database session for data retrieval
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.verify_link(payload, signature)
+
+ if link_data.get("link_type") != "redirect":
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ "This link is not a redirect link",
+ )
+
+ return _get_hitl_detail(
+ dag_id=link_data["dag_id"],
+ dag_run_id=link_data["dag_run_id"],
+ task_id=link_data["task_id"],
+ session=session,
+ map_index=link_data.get("map_index"),
+ )
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.get(
+
"/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}/{map_index}",
+ status_code=status.HTTP_200_OK,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+)
+def get_mapped_ti_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int,
+ payload: str,
+ signature: str,
+ session: SessionDep,
+) -> HITLDetail:
+ """
+ Get mapped HITL details via shared link (for redirect links).
+
+ This endpoint allows external users to access mapped HITL task details
through a secure
+ shared link. The link must be a redirect-type link, which provides
read-only access
+ to the mapped HITL task information for UI rendering or decision-making
purposes.
+
+ :param dag_id: The DAG identifier (from URL path)
+ :param dag_run_id: The DAG run identifier (from URL path)
+ :param task_id: The task identifier (from URL path)
+ :param map_index: The map index for the mapped task instance (from URL
path)
+ :param payload: Base64-encoded payload containing link metadata and
expiration
+ :param signature: HMAC signature for payload verification
+ :param session: Database session for data retrieval
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.verify_link(payload, signature)
+
+ if link_data.get("link_type") != "redirect":
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ "This link is not a redirect link",
+ )
+
+ return _get_hitl_detail(
+ dag_id=link_data["dag_id"],
+ dag_run_id=link_data["dag_run_id"],
+ task_id=link_data["task_id"],
+ session=session,
+ map_index=link_data.get("map_index"),
+ )
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.post(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}/action",
+ status_code=status.HTTP_200_OK,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+)
+def execute_hitl_share_link_action(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ payload: str,
+ signature: str,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Execute an action via shared link (for action links).
+
+ This endpoint allows external users to execute HITL task actions through a
secure
+ shared link. The link must be an action-type link, which enables direct
execution
+ of predefined actions (e.g., approve, reject) without requiring full
Airflow
+ authentication. The action is executed immediately and the HITL task is
updated
+ with the user's response.
+
+ :param dag_id: The DAG identifier (from URL path)
+ :param dag_run_id: The DAG run identifier (from URL path)
+ :param task_id: The task identifier (from URL path)
+ :param payload: Base64-encoded payload containing link metadata and
expiration
+ :param signature: HMAC signature for payload verification
+ :param update_hitl_detail_payload: Payload containing the action response
data
+ :param session: Database session for data persistence
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.verify_link(payload, signature)
+
+ if link_data.get("link_type") != "action":
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ "This link is not an action link",
+ )
+
+ return _update_hitl_detail(
+ dag_id=link_data["dag_id"],
+ dag_run_id=link_data["dag_run_id"],
+ task_id=link_data["task_id"],
+ session=session,
+ update_hitl_detail_payload=update_hitl_detail_payload,
+ user=None,
+ map_index=link_data.get("map_index"),
+ )
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.post(
+
"/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}/{map_index}/action",
+ status_code=status.HTTP_200_OK,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+)
+def execute_mapped_ti_hitl_share_link_action(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int,
+ payload: str,
+ signature: str,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Execute an action via shared link for mapped tasks (for action links).
+
+ This endpoint allows external users to execute mapped HITL task actions
through a secure
+ shared link. The link must be an action-type link, which enables direct
execution
+ of predefined actions (e.g., approve, reject) for specific mapped task
instances
+ without requiring full Airflow authentication. The action is executed
immediately
+ and the mapped HITL task is updated with the user's response.
+
+ :param dag_id: The DAG identifier (from URL path)
+ :param dag_run_id: The DAG run identifier (from URL path)
+ :param task_id: The task identifier (from URL path)
+ :param map_index: The map index for the mapped task instance (from URL
path)
+ :param payload: Base64-encoded payload containing link metadata and
expiration
+ :param signature: HMAC signature for payload verification
+ :param update_hitl_detail_payload: Payload containing the action response
data
+ :param session: Database session for data persistence
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.verify_link(payload, signature)
+
+ if link_data.get("link_type") != "action":
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ "This link is not an action link",
+ )
+
+ return _update_hitl_detail(
+ dag_id=link_data["dag_id"],
+ dag_run_id=link_data["dag_run_id"],
+ task_id=link_data["task_id"],
+ session=session,
+ update_hitl_detail_payload=update_hitl_detail_payload,
+ user=None,
+ map_index=link_data.get("map_index"),
Review Comment:
Same issue as the previous one for `map_index`.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py:
##########
@@ -272,3 +293,456 @@ def get_hitl_details(
hitl_details=hitl_details,
total_entries=total_entries,
)
+
+
+@hitl_router.post(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}",
+ status_code=status.HTTP_201_CREATED,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+ dependencies=[Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.TASK_INSTANCE))],
+)
+def create_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ user: GetUserDep,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Create a shared link for a Human-in-the-loop task.
+
+ This endpoint generates a secure, time-limited shared link that allows
external users
+ to interact with HITL tasks without requiring full Airflow authentication.
The link
+ can be configured for either direct action execution or UI redirection.
+
+ :param dag_id: The DAG identifier
+ :param dag_run_id: The DAG run identifier
+ :param task_id: The task identifier
+ :param update_hitl_detail_payload: Payload containing link configuration
and initial response data
+ :param user: The authenticated user creating the shared link
+ :param session: Database session for data persistence
+
+ :raises HTTPException: 403 if HITL shared links are not enabled
+ :raises HTTPException: 404 if the task instance or HITL detail does not
exist
+ :raises HTTPException: 400 if link generation fails due to invalid
parameters
+
+ :return: HITLDetailResponse containing the generated link URL and metadata
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ task_instance = _get_task_instance(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ session=session,
+ map_index=None,
+ )
+
+ ti_id_str = str(task_instance.id)
+ hitl_detail_model =
session.scalar(select(HITLDetailModel).where(HITLDetailModel.ti_id ==
ti_id_str))
+ if not hitl_detail_model:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Human-in-the-loop detail does not exist for Task Instance with
id {ti_id_str}",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.generate_link(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ map_index=None,
+ link_type=update_hitl_detail_payload.link_type,
+ action=update_hitl_detail_payload.action,
+ expires_in_hours=update_hitl_detail_payload.expires_in_hours,
+ )
+
+ response = HITLDetailResponse(
+ user_id=user.get_id(),
+ response_at=timezone.utcnow(),
+ chosen_options=update_hitl_detail_payload.chosen_options,
+ params_input=update_hitl_detail_payload.params_input,
+ task_instance_id=link_data["task_instance_id"],
+ link_url=link_data["link_url"],
+ expires_at=link_data["expires_at"],
+ action=link_data["action"],
+ link_type=link_data["link_type"],
+ )
+
+ return response
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.post(
+
"/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}/{map_index}",
+ status_code=status.HTTP_201_CREATED,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+ dependencies=[Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.TASK_INSTANCE))],
+)
+def create_mapped_ti_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ user: GetUserDep,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Create a shared link for a mapped Human-in-the-loop task.
+
+ This endpoint generates a secure, time-limited shared link for mapped task
instances,
+ allowing external users to interact with specific mapped HITL tasks
without requiring
+ full Airflow authentication. The link can be configured for either direct
action
+ execution or UI redirection.
+
+ :param dag_id: The DAG identifier
+ :param dag_run_id: The DAG run identifier
+ :param task_id: The task identifier
+ :param map_index: The map index for the mapped task instance
+ :param update_hitl_detail_payload: Payload containing link configuration
and initial response data
+ :param user: The authenticated user creating the shared link
+ :param session: Database session for data persistence
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ task_instance = _get_task_instance(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ session=session,
+ map_index=map_index,
+ )
+
+ ti_id_str = str(task_instance.id)
+ hitl_detail_model =
session.scalar(select(HITLDetailModel).where(HITLDetailModel.ti_id ==
ti_id_str))
+ if not hitl_detail_model:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Human-in-the-loop detail does not exist for Task Instance with
id {ti_id_str}",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.generate_link(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ map_index=map_index,
+ link_type=update_hitl_detail_payload.link_type,
+ action=update_hitl_detail_payload.action,
+ expires_in_hours=update_hitl_detail_payload.expires_in_hours,
+ )
+
+ response = HITLDetailResponse(
+ user_id=user.get_id(),
+ response_at=timezone.utcnow(),
+ chosen_options=update_hitl_detail_payload.chosen_options,
+ params_input=update_hitl_detail_payload.params_input,
+ task_instance_id=link_data["task_instance_id"],
+ link_url=link_data["link_url"],
+ expires_at=link_data["expires_at"],
+ action=link_data["action"],
+ link_type=link_data["link_type"],
+ )
+
+ return response
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.get(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}",
+ status_code=status.HTTP_200_OK,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+)
+def get_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ payload: str,
+ signature: str,
+ session: SessionDep,
+) -> HITLDetail:
+ """
+ Get HITL details via shared link (for redirect links).
+
+ This endpoint allows external users to access HITL task details through a
secure
+ shared link. The link must be a redirect-type link, which provides
read-only access
+ to the HITL task information for UI rendering or decision-making purposes.
+
+ :param dag_id: The DAG identifier (from URL path)
+ :param dag_run_id: The DAG run identifier (from URL path)
+ :param task_id: The task identifier (from URL path)
+ :param payload: Base64-encoded payload containing link metadata and
expiration
+ :param signature: HMAC signature for payload verification
+ :param session: Database session for data retrieval
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.verify_link(payload, signature)
+
+ if link_data.get("link_type") != "redirect":
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ "This link is not a redirect link",
+ )
+
+ return _get_hitl_detail(
+ dag_id=link_data["dag_id"],
+ dag_run_id=link_data["dag_run_id"],
+ task_id=link_data["task_id"],
+ session=session,
+ map_index=link_data.get("map_index"),
+ )
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.get(
+
"/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}/{map_index}",
+ status_code=status.HTTP_200_OK,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+)
+def get_mapped_ti_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int,
+ payload: str,
+ signature: str,
+ session: SessionDep,
+) -> HITLDetail:
+ """
+ Get mapped HITL details via shared link (for redirect links).
+
+ This endpoint allows external users to access mapped HITL task details
through a secure
+ shared link. The link must be a redirect-type link, which provides
read-only access
+ to the mapped HITL task information for UI rendering or decision-making
purposes.
+
+ :param dag_id: The DAG identifier (from URL path)
+ :param dag_run_id: The DAG run identifier (from URL path)
+ :param task_id: The task identifier (from URL path)
+ :param map_index: The map index for the mapped task instance (from URL
path)
+ :param payload: Base64-encoded payload containing link metadata and
expiration
+ :param signature: HMAC signature for payload verification
+ :param session: Database session for data retrieval
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.verify_link(payload, signature)
+
+ if link_data.get("link_type") != "redirect":
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ "This link is not a redirect link",
+ )
+
+ return _get_hitl_detail(
+ dag_id=link_data["dag_id"],
+ dag_run_id=link_data["dag_run_id"],
+ task_id=link_data["task_id"],
+ session=session,
+ map_index=link_data.get("map_index"),
+ )
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.post(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}/action",
+ status_code=status.HTTP_200_OK,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+)
+def execute_hitl_share_link_action(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ payload: str,
+ signature: str,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Execute an action via shared link (for action links).
+
+ This endpoint allows external users to execute HITL task actions through a
secure
+ shared link. The link must be an action-type link, which enables direct
execution
+ of predefined actions (e.g., approve, reject) without requiring full
Airflow
+ authentication. The action is executed immediately and the HITL task is
updated
+ with the user's response.
+
+ :param dag_id: The DAG identifier (from URL path)
+ :param dag_run_id: The DAG run identifier (from URL path)
+ :param task_id: The task identifier (from URL path)
+ :param payload: Base64-encoded payload containing link metadata and
expiration
+ :param signature: HMAC signature for payload verification
+ :param update_hitl_detail_payload: Payload containing the action response
data
+ :param session: Database session for data persistence
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.verify_link(payload, signature)
+
+ if link_data.get("link_type") != "action":
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ "This link is not an action link",
+ )
+
+ return _update_hitl_detail(
+ dag_id=link_data["dag_id"],
+ dag_run_id=link_data["dag_run_id"],
+ task_id=link_data["task_id"],
+ session=session,
+ update_hitl_detail_payload=update_hitl_detail_payload,
+ user=None,
+ map_index=link_data.get("map_index"),
+ )
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.post(
+
"/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}/{map_index}/action",
+ status_code=status.HTTP_200_OK,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+)
+def execute_mapped_ti_hitl_share_link_action(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int,
+ payload: str,
+ signature: str,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Execute an action via shared link for mapped tasks (for action links).
+
+ This endpoint allows external users to execute mapped HITL task actions
through a secure
+ shared link. The link must be an action-type link, which enables direct
execution
+ of predefined actions (e.g., approve, reject) for specific mapped task
instances
+ without requiring full Airflow authentication. The action is executed
immediately
+ and the mapped HITL task is updated with the user's response.
+
+ :param dag_id: The DAG identifier (from URL path)
+ :param dag_run_id: The DAG run identifier (from URL path)
+ :param task_id: The task identifier (from URL path)
+ :param map_index: The map index for the mapped task instance (from URL
path)
+ :param payload: Base64-encoded payload containing link metadata and
expiration
+ :param signature: HMAC signature for payload verification
+ :param update_hitl_detail_payload: Payload containing the action response
data
+ :param session: Database session for data persistence
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.verify_link(payload, signature)
+
+ if link_data.get("link_type") != "action":
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ "This link is not an action link",
+ )
+
+ return _update_hitl_detail(
+ dag_id=link_data["dag_id"],
+ dag_run_id=link_data["dag_run_id"],
+ task_id=link_data["task_id"],
+ session=session,
+ update_hitl_detail_payload=update_hitl_detail_payload,
+ user=None,
+ map_index=link_data.get("map_index"),
+ )
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
Review Comment:
How about having common functions in `service` module for
- `create_hitl_share_link` and `create_mapped_hitl_share_link`
- `get_hitl_share_link` and `get_mapped_ti_hitl_share_link`
- `execute_hitl_share_link_action` and
`execute_mapped_ti_hitl_share_link_action`
the logic for simple one and mapped one seem the same.
Only the parameters of the routes are different.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py:
##########
@@ -272,3 +293,456 @@ def get_hitl_details(
hitl_details=hitl_details,
total_entries=total_entries,
)
+
+
+@hitl_router.post(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}",
+ status_code=status.HTTP_201_CREATED,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+ dependencies=[Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.TASK_INSTANCE))],
+)
+def create_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ user: GetUserDep,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Create a shared link for a Human-in-the-loop task.
+
+ This endpoint generates a secure, time-limited shared link that allows
external users
+ to interact with HITL tasks without requiring full Airflow authentication.
The link
+ can be configured for either direct action execution or UI redirection.
+
+ :param dag_id: The DAG identifier
+ :param dag_run_id: The DAG run identifier
+ :param task_id: The task identifier
+ :param update_hitl_detail_payload: Payload containing link configuration
and initial response data
+ :param user: The authenticated user creating the shared link
+ :param session: Database session for data persistence
+
+ :raises HTTPException: 403 if HITL shared links are not enabled
+ :raises HTTPException: 404 if the task instance or HITL detail does not
exist
+ :raises HTTPException: 400 if link generation fails due to invalid
parameters
+
+ :return: HITLDetailResponse containing the generated link URL and metadata
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ task_instance = _get_task_instance(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ session=session,
+ map_index=None,
+ )
+
+ ti_id_str = str(task_instance.id)
+ hitl_detail_model =
session.scalar(select(HITLDetailModel).where(HITLDetailModel.ti_id ==
ti_id_str))
+ if not hitl_detail_model:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Human-in-the-loop detail does not exist for Task Instance with
id {ti_id_str}",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.generate_link(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ map_index=None,
+ link_type=update_hitl_detail_payload.link_type,
+ action=update_hitl_detail_payload.action,
+ expires_in_hours=update_hitl_detail_payload.expires_in_hours,
+ )
+
+ response = HITLDetailResponse(
+ user_id=user.get_id(),
+ response_at=timezone.utcnow(),
+ chosen_options=update_hitl_detail_payload.chosen_options,
+ params_input=update_hitl_detail_payload.params_input,
+ task_instance_id=link_data["task_instance_id"],
+ link_url=link_data["link_url"],
+ expires_at=link_data["expires_at"],
+ action=link_data["action"],
+ link_type=link_data["link_type"],
+ )
+
+ return response
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.post(
+
"/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}/{map_index}",
+ status_code=status.HTTP_201_CREATED,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+ dependencies=[Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.TASK_INSTANCE))],
+)
+def create_mapped_ti_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ user: GetUserDep,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Create a shared link for a mapped Human-in-the-loop task.
+
+ This endpoint generates a secure, time-limited shared link for mapped task
instances,
+ allowing external users to interact with specific mapped HITL tasks
without requiring
+ full Airflow authentication. The link can be configured for either direct
action
+ execution or UI redirection.
+
+ :param dag_id: The DAG identifier
+ :param dag_run_id: The DAG run identifier
+ :param task_id: The task identifier
+ :param map_index: The map index for the mapped task instance
+ :param update_hitl_detail_payload: Payload containing link configuration
and initial response data
+ :param user: The authenticated user creating the shared link
+ :param session: Database session for data persistence
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ task_instance = _get_task_instance(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ session=session,
+ map_index=map_index,
+ )
+
+ ti_id_str = str(task_instance.id)
+ hitl_detail_model =
session.scalar(select(HITLDetailModel).where(HITLDetailModel.ti_id ==
ti_id_str))
+ if not hitl_detail_model:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Human-in-the-loop detail does not exist for Task Instance with
id {ti_id_str}",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.generate_link(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ map_index=map_index,
+ link_type=update_hitl_detail_payload.link_type,
+ action=update_hitl_detail_payload.action,
+ expires_in_hours=update_hitl_detail_payload.expires_in_hours,
+ )
+
+ response = HITLDetailResponse(
+ user_id=user.get_id(),
+ response_at=timezone.utcnow(),
+ chosen_options=update_hitl_detail_payload.chosen_options,
+ params_input=update_hitl_detail_payload.params_input,
+ task_instance_id=link_data["task_instance_id"],
+ link_url=link_data["link_url"],
+ expires_at=link_data["expires_at"],
+ action=link_data["action"],
+ link_type=link_data["link_type"],
+ )
+
+ return response
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.get(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}",
+ status_code=status.HTTP_200_OK,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+)
+def get_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ payload: str,
+ signature: str,
+ session: SessionDep,
+) -> HITLDetail:
+ """
+ Get HITL details via shared link (for redirect links).
+
+ This endpoint allows external users to access HITL task details through a
secure
+ shared link. The link must be a redirect-type link, which provides
read-only access
+ to the HITL task information for UI rendering or decision-making purposes.
+
+ :param dag_id: The DAG identifier (from URL path)
+ :param dag_run_id: The DAG run identifier (from URL path)
+ :param task_id: The task identifier (from URL path)
+ :param payload: Base64-encoded payload containing link metadata and
expiration
+ :param signature: HMAC signature for payload verification
+ :param session: Database session for data retrieval
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.verify_link(payload, signature)
+
+ if link_data.get("link_type") != "redirect":
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ "This link is not a redirect link",
+ )
+
+ return _get_hitl_detail(
+ dag_id=link_data["dag_id"],
+ dag_run_id=link_data["dag_run_id"],
+ task_id=link_data["task_id"],
+ session=session,
+ map_index=link_data.get("map_index"),
+ )
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.get(
+
"/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}/{map_index}",
+ status_code=status.HTTP_200_OK,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
Review Comment:
```suggestion
responses=create_openapi_http_exception_doc(
[
status.HTTP_400_BAD_REQUEST,
status.HTTP_403_FORBIDDEN,
status.HTTP_404_NOT_FOUND,
]
),
```
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py:
##########
@@ -272,3 +293,456 @@ def get_hitl_details(
hitl_details=hitl_details,
total_entries=total_entries,
)
+
+
+@hitl_router.post(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}",
+ status_code=status.HTTP_201_CREATED,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+ dependencies=[Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.TASK_INSTANCE))],
+)
+def create_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ user: GetUserDep,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Create a shared link for a Human-in-the-loop task.
+
+ This endpoint generates a secure, time-limited shared link that allows
external users
+ to interact with HITL tasks without requiring full Airflow authentication.
The link
+ can be configured for either direct action execution or UI redirection.
+
+ :param dag_id: The DAG identifier
+ :param dag_run_id: The DAG run identifier
+ :param task_id: The task identifier
+ :param update_hitl_detail_payload: Payload containing link configuration
and initial response data
+ :param user: The authenticated user creating the shared link
+ :param session: Database session for data persistence
+
+ :raises HTTPException: 403 if HITL shared links are not enabled
+ :raises HTTPException: 404 if the task instance or HITL detail does not
exist
+ :raises HTTPException: 400 if link generation fails due to invalid
parameters
+
+ :return: HITLDetailResponse containing the generated link URL and metadata
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
Review Comment:
How about make the `hitl_shared_link_manager.is_enabled` as a `Depends` in
FastAPI?
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py:
##########
@@ -272,3 +293,456 @@ def get_hitl_details(
hitl_details=hitl_details,
total_entries=total_entries,
)
+
+
+@hitl_router.post(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}",
+ status_code=status.HTTP_201_CREATED,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+ dependencies=[Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.TASK_INSTANCE))],
+)
+def create_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ user: GetUserDep,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Create a shared link for a Human-in-the-loop task.
+
+ This endpoint generates a secure, time-limited shared link that allows
external users
+ to interact with HITL tasks without requiring full Airflow authentication.
The link
+ can be configured for either direct action execution or UI redirection.
+
+ :param dag_id: The DAG identifier
+ :param dag_run_id: The DAG run identifier
+ :param task_id: The task identifier
+ :param update_hitl_detail_payload: Payload containing link configuration
and initial response data
+ :param user: The authenticated user creating the shared link
+ :param session: Database session for data persistence
+
+ :raises HTTPException: 403 if HITL shared links are not enabled
+ :raises HTTPException: 404 if the task instance or HITL detail does not
exist
+ :raises HTTPException: 400 if link generation fails due to invalid
parameters
+
+ :return: HITLDetailResponse containing the generated link URL and metadata
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ task_instance = _get_task_instance(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ session=session,
+ map_index=None,
+ )
+
+ ti_id_str = str(task_instance.id)
+ hitl_detail_model =
session.scalar(select(HITLDetailModel).where(HITLDetailModel.ti_id ==
ti_id_str))
+ if not hitl_detail_model:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Human-in-the-loop detail does not exist for Task Instance with
id {ti_id_str}",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.generate_link(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ map_index=None,
+ link_type=update_hitl_detail_payload.link_type,
+ action=update_hitl_detail_payload.action,
+ expires_in_hours=update_hitl_detail_payload.expires_in_hours,
+ )
+
+ response = HITLDetailResponse(
+ user_id=user.get_id(),
+ response_at=timezone.utcnow(),
+ chosen_options=update_hitl_detail_payload.chosen_options,
+ params_input=update_hitl_detail_payload.params_input,
+ task_instance_id=link_data["task_instance_id"],
+ link_url=link_data["link_url"],
+ expires_at=link_data["expires_at"],
+ action=link_data["action"],
+ link_type=link_data["link_type"],
+ )
+
+ return response
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.post(
+
"/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}/{map_index}",
+ status_code=status.HTTP_201_CREATED,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+ dependencies=[Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.TASK_INSTANCE))],
+)
+def create_mapped_ti_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int,
+ update_hitl_detail_payload: UpdateHITLDetailPayload,
+ user: GetUserDep,
+ session: SessionDep,
+) -> HITLDetailResponse:
+ """
+ Create a shared link for a mapped Human-in-the-loop task.
+
+ This endpoint generates a secure, time-limited shared link for mapped task
instances,
+ allowing external users to interact with specific mapped HITL tasks
without requiring
+ full Airflow authentication. The link can be configured for either direct
action
+ execution or UI redirection.
+
+ :param dag_id: The DAG identifier
+ :param dag_run_id: The DAG run identifier
+ :param task_id: The task identifier
+ :param map_index: The map index for the mapped task instance
+ :param update_hitl_detail_payload: Payload containing link configuration
and initial response data
+ :param user: The authenticated user creating the shared link
+ :param session: Database session for data persistence
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ task_instance = _get_task_instance(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ session=session,
+ map_index=map_index,
+ )
+
+ ti_id_str = str(task_instance.id)
+ hitl_detail_model =
session.scalar(select(HITLDetailModel).where(HITLDetailModel.ti_id ==
ti_id_str))
+ if not hitl_detail_model:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Human-in-the-loop detail does not exist for Task Instance with
id {ti_id_str}",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.generate_link(
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ task_id=task_id,
+ map_index=map_index,
+ link_type=update_hitl_detail_payload.link_type,
+ action=update_hitl_detail_payload.action,
+ expires_in_hours=update_hitl_detail_payload.expires_in_hours,
+ )
+
+ response = HITLDetailResponse(
+ user_id=user.get_id(),
+ response_at=timezone.utcnow(),
+ chosen_options=update_hitl_detail_payload.chosen_options,
+ params_input=update_hitl_detail_payload.params_input,
+ task_instance_id=link_data["task_instance_id"],
+ link_url=link_data["link_url"],
+ expires_at=link_data["expires_at"],
+ action=link_data["action"],
+ link_type=link_data["link_type"],
+ )
+
+ return response
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.get(
+ "/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}",
+ status_code=status.HTTP_200_OK,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+)
+def get_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ payload: str,
+ signature: str,
+ session: SessionDep,
+) -> HITLDetail:
+ """
+ Get HITL details via shared link (for redirect links).
+
+ This endpoint allows external users to access HITL task details through a
secure
+ shared link. The link must be a redirect-type link, which provides
read-only access
+ to the HITL task information for UI rendering or decision-making purposes.
+
+ :param dag_id: The DAG identifier (from URL path)
+ :param dag_run_id: The DAG run identifier (from URL path)
+ :param task_id: The task identifier (from URL path)
+ :param payload: Base64-encoded payload containing link metadata and
expiration
+ :param signature: HMAC signature for payload verification
+ :param session: Database session for data retrieval
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.verify_link(payload, signature)
+
+ if link_data.get("link_type") != "redirect":
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ "This link is not a redirect link",
+ )
+
+ return _get_hitl_detail(
+ dag_id=link_data["dag_id"],
+ dag_run_id=link_data["dag_run_id"],
+ task_id=link_data["task_id"],
+ session=session,
+ map_index=link_data.get("map_index"),
+ )
+
+ except ValueError as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ str(e),
+ )
+
+
+@hitl_router.get(
+
"/api/v2/hitl-details-share-link/{dag_id}/{dag_run_id}/{task_id}/{map_index}",
+ status_code=status.HTTP_200_OK,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_403_FORBIDDEN,
+ ]
+ ),
+)
+def get_mapped_ti_hitl_share_link(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int,
+ payload: str,
+ signature: str,
+ session: SessionDep,
+) -> HITLDetail:
+ """
+ Get mapped HITL details via shared link (for redirect links).
+
+ This endpoint allows external users to access mapped HITL task details
through a secure
+ shared link. The link must be a redirect-type link, which provides
read-only access
+ to the mapped HITL task information for UI rendering or decision-making
purposes.
+
+ :param dag_id: The DAG identifier (from URL path)
+ :param dag_run_id: The DAG run identifier (from URL path)
+ :param task_id: The task identifier (from URL path)
+ :param map_index: The map index for the mapped task instance (from URL
path)
+ :param payload: Base64-encoded payload containing link metadata and
expiration
+ :param signature: HMAC signature for payload verification
+ :param session: Database session for data retrieval
+ """
+ if not hitl_shared_link_manager.is_enabled():
+ raise HTTPException(
+ status.HTTP_403_FORBIDDEN,
+ "HITL shared links are not enabled",
+ )
+
+ try:
+ link_data = hitl_shared_link_manager.verify_link(payload, signature)
+
+ if link_data.get("link_type") != "redirect":
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ "This link is not a redirect link",
+ )
+
+ return _get_hitl_detail(
+ dag_id=link_data["dag_id"],
+ dag_run_id=link_data["dag_run_id"],
+ task_id=link_data["task_id"],
+ session=session,
+ map_index=link_data.get("map_index"),
Review Comment:
Is seem the `map_index` from route is not used in handler function at all.
Should it be `map_index= map_index` here ?
##########
airflow-core/src/airflow/utils/hitl_shared_links.py:
##########
@@ -0,0 +1,220 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Utilities for Human-in-the-Loop (HITL) shared links."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+from datetime import datetime, timedelta
+from typing import Any
+from urllib.parse import urlencode
+
+import structlog
+
+from airflow.configuration import conf
+from airflow.utils import timezone
+
+log = structlog.get_logger(__name__)
+
+
+class HITLSharedLinkManager:
+ """Manager for HITL shared links with token generation and verification."""
+
+ def __init__(self):
+ self.secret_key = conf.get("api", "hitl_shared_link_secret_key",
fallback="")
+ self.default_expiration_hours = conf.getint("api",
"hitl_shared_link_expiration_hours", fallback=24)
+
+ def is_enabled(self) -> bool:
+ """Check if HITL shared links are enabled."""
+ return conf.getboolean("api", "hitl_enable_shared_links",
fallback=False)
+
+ def _generate_signature(self, payload: str) -> str:
+ """Generate HMAC signature for the payload."""
+ if not self.secret_key:
+ raise ValueError("HITL shared link secret key is not configured")
+
+ signature = hmac.new(
+ self.secret_key.encode("utf-8"), payload.encode("utf-8"),
hashlib.sha256
+ ).digest()
+ return base64.urlsafe_b64encode(signature).decode("utf-8")
+
+ def _verify_signature(self, payload: str, signature: str) -> bool:
+ """Verify HMAC signature for the payload."""
+ expected_signature = self._generate_signature(payload)
+ return hmac.compare_digest(expected_signature, signature)
+
+ def generate_link(
+ self,
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int | None = None,
+ link_type: str = "action",
+ action: str | None = None,
+ expires_in_hours: int | None = None,
+ base_url: str | None = None,
+ ) -> dict[str, Any]:
+ """
+ Generate a shared link for HITL task.
+
+ :param dag_id: DAG ID
+ :param dag_run_id: DAG run ID
+ :param task_id: Task ID
+ :param map_index: Map index for mapped tasks
+ :param link_type: Type of link ('action' or 'redirect')
+ :param action: Action to perform (for action links)
+ :param expires_in_hours: Custom expiration time in hours
+ :param base_url: Base URL for the link
+ """
+ if not self.is_enabled():
+ raise ValueError("HITL shared links are not enabled")
+
+ if link_type == "action" and not action:
+ raise ValueError("Action is required for action-type links")
+
+ expiration_hours = expires_in_hours or self.default_expiration_hours
+ expires_at = timezone.utcnow() + timedelta(hours=expiration_hours)
+
+ payload_data = {
+ "dag_id": dag_id,
+ "dag_run_id": dag_run_id,
+ "task_id": task_id,
+ "map_index": map_index,
+ "link_type": link_type,
+ "action": action,
+ "expires_at": expires_at.isoformat(),
+ }
Review Comment:
Or maybe pydantic model will be better as pydantic will validate the
required payload when load from the `payload_str`.
##########
airflow-core/src/airflow/utils/hitl_shared_links.py:
##########
@@ -0,0 +1,220 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Utilities for Human-in-the-Loop (HITL) shared links."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+from datetime import datetime, timedelta
+from typing import Any
+from urllib.parse import urlencode
+
+import structlog
+
+from airflow.configuration import conf
+from airflow.utils import timezone
+
+log = structlog.get_logger(__name__)
+
+
+class HITLSharedLinkManager:
+ """Manager for HITL shared links with token generation and verification."""
+
+ def __init__(self):
+ self.secret_key = conf.get("api", "hitl_shared_link_secret_key",
fallback="")
+ self.default_expiration_hours = conf.getint("api",
"hitl_shared_link_expiration_hours", fallback=24)
+
+ def is_enabled(self) -> bool:
+ """Check if HITL shared links are enabled."""
+ return conf.getboolean("api", "hitl_enable_shared_links",
fallback=False)
+
+ def _generate_signature(self, payload: str) -> str:
+ """Generate HMAC signature for the payload."""
+ if not self.secret_key:
+ raise ValueError("HITL shared link secret key is not configured")
+
+ signature = hmac.new(
+ self.secret_key.encode("utf-8"), payload.encode("utf-8"),
hashlib.sha256
+ ).digest()
+ return base64.urlsafe_b64encode(signature).decode("utf-8")
+
+ def _verify_signature(self, payload: str, signature: str) -> bool:
+ """Verify HMAC signature for the payload."""
+ expected_signature = self._generate_signature(payload)
+ return hmac.compare_digest(expected_signature, signature)
+
+ def generate_link(
+ self,
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int | None = None,
+ link_type: str = "action",
+ action: str | None = None,
+ expires_in_hours: int | None = None,
+ base_url: str | None = None,
+ ) -> dict[str, Any]:
+ """
+ Generate a shared link for HITL task.
+
+ :param dag_id: DAG ID
+ :param dag_run_id: DAG run ID
+ :param task_id: Task ID
+ :param map_index: Map index for mapped tasks
+ :param link_type: Type of link ('action' or 'redirect')
+ :param action: Action to perform (for action links)
+ :param expires_in_hours: Custom expiration time in hours
+ :param base_url: Base URL for the link
+ """
+ if not self.is_enabled():
+ raise ValueError("HITL shared links are not enabled")
+
+ if link_type == "action" and not action:
+ raise ValueError("Action is required for action-type links")
+
+ expiration_hours = expires_in_hours or self.default_expiration_hours
+ expires_at = timezone.utcnow() + timedelta(hours=expiration_hours)
+
+ payload_data = {
+ "dag_id": dag_id,
+ "dag_run_id": dag_run_id,
+ "task_id": task_id,
+ "map_index": map_index,
+ "link_type": link_type,
+ "action": action,
+ "expires_at": expires_at.isoformat(),
+ }
+
+ payload_str = json.dumps(payload_data, sort_keys=True)
+ signature = self._generate_signature(payload_str)
+
+ encoded_payload =
base64.urlsafe_b64encode(payload_str.encode("utf-8")).decode("utf-8")
+
+ if base_url is None:
+ base_url = conf.get("api", "base_url",
fallback="http://localhost:8080";)
Review Comment:
IMO, this fallback is same as `TaskInstance.log_url`
https://github.com/apache/airflow/blob/67f204417b0afd3e3b8713121f2b171564c1909a/airflow-core/src/airflow/models/taskinstance.py#L918-L923
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
