mirror of
https://github.com/apache/airflow.git
synced 2026-11-04 07:12:48 +00:00
ZachandAsh Berlin-Taylor
63f5845dd2
* Depend on httpx2 instead of httpx in the Task SDK httpx has stalled: fixes that landed in httpx2 (a fork of httpx 0.28.1) have not shipped upstream, and starlette has already migrated, so FastAPI following is a matter of time. Moving the Execution API client now keeps the SDK on a maintained HTTP stack. Objects do not cross the httpx/httpx2 boundary, so the in-process Execution API transports move with the client that consumes them; the rest of airflow-core stays on httpx and migrates separately. related: https://github.com/apache/airflow/issues/70522 * Add newsfragment for the Task SDK httpx2 migration * Match the in-process test client transport to the installed Task SDK Compat tests pair the repo's devel-common helpers with a released Airflow whose Task SDK is still on httpx. Pinning the helper to httpx2 hands those clients a transport from the wrong package, and httpx rejects the response stream it gets back. * Rename newsfragment to the new PR number * Pin the Task SDK ast-grep hook to a frozen revision Every other third-party hook in the repo pins an immutable SHA; a mutable tag lets the hook change under CI without a diff. * Remove unneeded newsfragment * Verify Task SDK HTTP logging and simplify test helpers Both HTTP stacks can coexist while compatibility jobs use older SDK distributions. The migration needs regression coverage for its logging behavior without exposing development-only helpers as public APIs. * Use httpx2 in the Dag run note client test * Use httpx2 in the execution API shutdown tests Tests added on main for in-process shutdown still referenced httpx, which this branch no longer imports in the module. * Trust the system certificate store in the Task SDK API client httpx2 verifies against the system trust store via truststore, so keeping certifi for the Execution API client would add a direct dependency only to maintain a second, different set of trusted roots in the same worker. * Ban httpx imports in the Task SDK with ruff Ruff already enforces banned imports, so a separate ast-grep hook and its extra pre-commit dependency are unnecessary. extend-banned-api adds to the root banned-api table instead of replacing it. * Use httpx2 in the server termination acknowledgement test The test arrived on main after the Task SDK moved to httpx2 and still referenced httpx, which the module no longer imports. * Test the Task SDK SSL context against a real CA file Mocking load_verify_locations while also writing a CA file to disk tested neither path convincingly. Keeping the certificate-generation imports inside the fixture limits them to the one place that needs them. * Note when the in-process httpx resolver can be removed * Apply suggestion from @ashb Co-authored-by: Ash Berlin-Taylor <ash_github@firemirror.com> * Use httpx2 in the supervisor launch error test helper The helper arrived on main after the Task SDK moved to httpx2 and still referenced httpx, which the module no longer imports. * Remove the duplicated in-process httpx resolver note * Use httpx2 in the supervisor execution timeout test The test arrived on main after the Task SDK moved to httpx2 and still referenced httpx, which the module no longer imports. * Use httpx2 in the request handler tests The tests arrived on main after the Task SDK moved to httpx2; the client rejects httpx transports. --------- Co-authored-by: Ash Berlin-Taylor <ash_github@firemirror.com>
96 lines
3.3 KiB
Python
96 lines
3.3 KiB
Python
# 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.
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from tomllib import load as load_tomllib
|
|
|
|
import httpx2
|
|
from datamodel_code_generator import (
|
|
DataModelType,
|
|
DatetimeClassType,
|
|
InputFileType,
|
|
LiteralType,
|
|
PythonVersion,
|
|
generate as generate_models,
|
|
)
|
|
from openapi_spec_validator import validate_spec
|
|
|
|
os.environ["_AIRFLOW__AS_LIBRARY"] = "1"
|
|
|
|
AIRFLOW_ROOT_PATH = Path(__file__).parents[2].resolve()
|
|
AIRFLOW_TASK_SDK_ROOT_PATH = AIRFLOW_ROOT_PATH / "task-sdk"
|
|
AIRFLOW_CORE_SOURCES_PATH = AIRFLOW_ROOT_PATH / "airflow-core" / "src"
|
|
sys.path.insert(0, str(Path(__file__).parent.resolve())) # make sure common_prek_utils is imported
|
|
|
|
sys.path.insert(0, AIRFLOW_CORE_SOURCES_PATH.as_posix())
|
|
sys.path.insert(0, str(AIRFLOW_ROOT_PATH)) # make sure setup is imported from Airflow
|
|
|
|
|
|
task_sdk_root = Path(__file__).parents[1]
|
|
|
|
|
|
def load_config():
|
|
pyproject = AIRFLOW_TASK_SDK_ROOT_PATH / "pyproject.toml"
|
|
# Simulate what `datamodel-code-generator` does on the CLI
|
|
with pyproject.open("rb") as fh:
|
|
cfg = load_tomllib(fh)["tool"]["datamodel-codegen"]
|
|
|
|
cfg = {k.replace("-", "_"): v for k, v in cfg.items()}
|
|
|
|
# Translate config file option names to generate() kwarg names
|
|
if (use_default := cfg.pop("use_default", None)) is not None:
|
|
cfg["apply_default_values_for_required_fields"] = use_default
|
|
|
|
if cfg.get("use_annotated"):
|
|
cfg["field_constraints"] = True
|
|
|
|
cfg["output"] = Path(cfg["output"])
|
|
cfg["output_model_type"] = DataModelType(cfg["output_model_type"])
|
|
cfg["output_datetime_class"] = DatetimeClassType(cfg["output_datetime_class"])
|
|
cfg["input_file_type"] = InputFileType(cfg["input_file_type"])
|
|
cfg["target_python_version"] = PythonVersion(cfg["target_python_version"])
|
|
cfg["enum_field_as_literal"] = LiteralType(cfg["enum_field_as_literal"])
|
|
return cfg
|
|
|
|
|
|
def generate_file():
|
|
from airflow.api_fastapi.execution_api.app import InProcessExecutionAPI
|
|
|
|
app = InProcessExecutionAPI()
|
|
|
|
latest_version = app.app.versions.version_values[0]
|
|
client = httpx2.Client(transport=app.transport)
|
|
openapi_schema = (
|
|
client.get(f"http://localhost/openapi.json?version={latest_version}").raise_for_status().text
|
|
)
|
|
|
|
validate_spec(json.loads(openapi_schema))
|
|
|
|
os.chdir(AIRFLOW_TASK_SDK_ROOT_PATH)
|
|
|
|
args = load_config()
|
|
args["input_filename"] = args.pop("url")
|
|
args["custom_formatters_kwargs"] = {"api_version": latest_version}
|
|
generate_models(openapi_schema, **args)
|
|
|
|
|
|
generate_file()
|