ashb commented on code in PR #69757: URL: https://github.com/apache/airflow/pull/69757#discussion_r3728080264
########## airflow-core/src/airflow/api_fastapi/execution_api/datamodels/task_arg_binding.py: ########## @@ -0,0 +1,112 @@ +# 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. +""" +Positional-argument binding spec for stub (foreign-runtime) tasks. + +Captured at parse time from a stub task's TaskFlow call (``@task.stub``), stored in the +serialized Dag, and delivered to the lang-SDK runtime through ``TIRunContext.arg_bindings`` +so it can bind the values onto the native task function's parameters. + +Each binding is one variant of a union discriminated on ``kind``: an ``XComArgBinding`` +pulls the value from an upstream task's XCom, a ``LiteralArgBinding`` carries an inline +value from the Dag file. Both variants still emit plain named structs +(``$defs/XComArgBinding``, ``$defs/LiteralArgBinding``) for the foreign-language SDKs +consuming the supervisor schema. +""" + +from __future__ import annotations + +from enum import Enum +from functools import cache +from typing import Annotated, Literal + +from pydantic import Field, JsonValue, TypeAdapter +from typing_extensions import TypeAliasType + +from airflow.api_fastapi.core_api.base import BaseModel + + +class ArgBindingDataType(str, Enum): + """Language-neutral value type a stub-task argument binds to in the foreign runtime.""" + + STRING = "string" + INTEGER = "integer" + NUMBER = "number" + BOOLEAN = "boolean" + OBJECT = "object" + ARRAY = "array" + ANY = "any" + + +class XComArgBinding(BaseModel): + """One positional stub-task argument pulled from an upstream task's XCom.""" + + # No default on purpose: a required ``kind`` stays non-nullable through the OpenAPI + # round trip, which discriminated-union codegen needs (a defaulted field turns + # ``Literal`` into ``Literal | None`` in the generated task-sdk models). + kind: Literal["xcom"] + + name: str + """The stub function's parameter name this binding fills, in declaration order.""" + + data_type: ArgBindingDataType = ArgBindingDataType.ANY + """Declared type from the stub function's annotation; runtimes type-check against it.""" + + task_id: str + """Upstream task id to pull the XCom from.""" + + key: str = "return_value" + """XCom key to pull.""" Review Comment: That isn't true. This example from https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/taskflow.html ```python @task def get_ip(): return my_ip_service.get_main_ip() @task(multiple_outputs=True) def compose_email(external_ip): return { 'subject':f'Server connected from {external_ip}', 'body': f'Your server executing Airflow is connected from the external IP {external_ip}<br>' } email_info = compose_email(get_ip()) EmailOperator( task_id='send_email_notification', to='[email protected]', subject=email_info['subject'], html_content=email_info['body'] ) ``` shows that compose_email actually stores two different xcom_keys, `subject` and `body`, and `XcomArg`'s subscripte operator gives different keys. -- 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]
