You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "chen0427ok (via GitHub)" <gi...@apache.org> on 2025/12/11 06:05:04 UTC
[PR] feat: Add dynamic parameter config source support for DAG params [airflow]
chen0427ok opened a new pull request, #59291:
URL: https://github.com/apache/airflow/pull/59291
## Description
Closes #56632
This PR adds support for loading DAG parameter options dynamically from external configuration files (INI, JSON, YAML) with filtering capabilities.
## Changes
### New Files
- **`airflow-core/src/airflow/utils/param_config_loader.py`**: Utility functions to load options from config files
- `load_options_from_ini()` - Load from INI files with filtering
- `load_options_from_json()` - Load from JSON files with filtering
- `load_options_from_yaml()` - Load from YAML files with filtering
- All support filter conditions and custom key fields
### Modified Files
- **`task-sdk/src/airflow/sdk/definitions/param.py`**: Extended `Param` class
- Added `config_source` parameter to `__init__()`
- Added `_load_options_from_config()` method to load options from external files
- Updated `__copy__()`, `serialize()`, and `deserialize()` methods
- Automatically populates `schema["enum"]` from config files at DAG parse time
### Test Files
- **`airflow-core/tests/unit/utils/test_param_config_loader.py`**: 18 unit tests for config loader
- **`task-sdk/tests/task_sdk/definitions/test_param.py`**: 4 new tests for Param class config_source feature
### Documentation
- **`airflow-core/newsfragments/56632.feature.rst`**: Release note for this feature
## Usage Example
**Config file** (`/path/to/interfaces.ini`):
```ini
[InterfaceA]
TYPE = Script
DESCRIPTION = Script based interface A
[InterfaceB]
TYPE = EBICS
DESCRIPTION = Electronic Banking interface
[InterfaceC]
TYPE = Script
DESCRIPTION = Another script interface
DAG definition:
from airflow import DAG
from airflow.sdk import Param
from datetime import datetime
with DAG(
dag_id="example_dynamic_params",
start_date=datetime(2024, 1, 1),
schedule=None,
params={
"interface_name": Param(
default="InterfaceA",
config_source={
"file": "/path/to/interfaces.ini",
"filter": {"TYPE": "Script"}, # Only show Script type interfaces
"key_field": "section",
},
description="Select an interface (dynamically loaded)",
),
},
) as dag:
# ... tasks
Result in UI:
- Renders a dropdown with only InterfaceA and InterfaceC (filtered by TYPE = Script)
- No frontend code changes needed - uses existing enum schema mechanism
Supported Config Formats
INI Files
- Filter by section properties
- Extract section names or specific field values
- Example: filter interfaces by type, environment, etc.
JSON Files
- Support array of objects
- Filter by object properties
- Extract specific field as option value
YAML Files
- Support array of objects
- Filter by object properties
- Extract specific field as option value
Testing
✅ All 22 unit tests passing:
- 18 tests for param_config_loader.py covering all formats and edge cases
- 4 tests for Param class config_source integration
✅ Manual testing in Airflow UI:
- Created test DAG with config_source parameter
- Verified dropdown rendering with filtered options
- Verified task execution with selected parameter values
- Tested with all three config formats (INI, JSON, YAML)
✅ Static checks:
- All prek hooks passed
- ruff formatting/linting passed
- mypy type checking passed
Test Commands
# Test config loader
uv run pytest airflow-core/tests/unit/utils/test_param_config_loader.py -v
# Test Param class integration
uv run pytest task-sdk/tests/task_sdk/definitions/test_param.py -k "config_source" -v
Implementation Notes
- Config files are loaded at DAG parse time, not at UI render time
- Options are stored in the enum field of the parameter's JSON schema
- If config loading fails (file not found, parsing error), a warning is logged and the parameter falls back to a regular string field
- No database schema changes required
- No UI code changes required - reuses existing enum dropdown rendering
- Backward compatible - existing DAGs without config_source continue to work unchanged
Follow-up Work
This PR provides the foundation for dynamic parameter options. Potential future enhancements:
- Support for database queries as config sources
- Support for API endpoints as config sources
- Caching mechanism for frequently accessed config files
- UI for managing config files
Checklist
- Feature implementation completed
- Unit tests added and passing (22/22)
- Manual testing in UI completed
- Static checks passing
- Newsfragment added
- Code follows Airflow coding standards
- Documentation in code (docstrings)
- Backward compatible
---
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add dynamic parameter config source support for DAG params [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #59291:
URL: https://github.com/apache/airflow/pull/59291#issuecomment-3881480832
This pull request has been closed because the author has not responded to a request for more information.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add dynamic parameter config source support for DAG params [airflow]
Posted by "chen0427ok (via GitHub)" <gi...@apache.org>.
chen0427ok commented on code in PR #59291:
URL: https://github.com/apache/airflow/pull/59291#discussion_r2611576942
##########
task-sdk/src/airflow/sdk/definitions/param.py:
##########
@@ -64,13 +81,67 @@ def __init__(
self.description = description
self.schema = kwargs.pop("schema") if "schema" in kwargs else kwargs
self.source = source
+ self.config_source = config_source
+
+ # If config_source is provided, load options and add to schema as enum
+ if config_source:
+ try:
+ options = self._load_options_from_config(config_source)
+ if options:
+ self.schema["enum"] = options
+ except Exception as e:
+ logger.warning(
+ "Failed to load options from config_source: %s. "
+ "Parameter will be treated as a regular string field.",
+ e,
+ )
def __copy__(self) -> Param:
return Param(
self.value,
self.description,
schema=self.schema,
source=self.source,
+ config_source=self.config_source,
+ )
+
+ def _load_options_from_config(self, config_source: dict[str, Any]) -> list[str]:
+ """
+ Load dropdown options from an external configuration file.
+
+ :param config_source: Dictionary containing configuration for loading options.
+ Must include 'file' key with path to config file.
+ Optional keys: 'filter' (dict), 'key_field' (str)
+ :return: List of option values loaded from the configuration file
+ :raises FileNotFoundError: If the configuration file is not found
+ :raises ValueError: If config_source is invalid or file format is unsupported
+ """
+ # Import here to avoid circular dependencies
+ from airflow.sdk.definitions.param_config_loader import (
+ load_options_from_ini,
+ load_options_from_json,
+ load_options_from_yaml,
+ )
+
+ file_path = config_source.get("file")
+ if not file_path:
+ raise ValueError("config_source must include 'file' key with path to configuration file")
+
+ filter_conditions = config_source.get("filter")
+ key_field = config_source.get("key_field")
+
+ # Detect file type and use appropriate loader
+ if file_path.endswith(".ini"):
+ key_field = key_field or "section"
+ return load_options_from_ini(file_path, filter_conditions, key_field)
+ if file_path.endswith(".json"):
+ key_field = key_field or "name"
+ return load_options_from_json(file_path, filter_conditions, key_field)
+ if file_path.endswith((".yaml", ".yml")):
+ key_field = key_field or "name"
+ return load_options_from_yaml(file_path, filter_conditions, key_field)
+ raise ValueError(
+ f"Unsupported config file type: {file_path}. Supported types are: .ini, .json, .yaml, .yml"
Review Comment:
Thanks for the detailed review! I've addressed all your feedback:
1. Consolidated the three loader functions into `load_options_from_file()` with an `extension` parameter
2. Added `ConfigSource` TypedDict for better type hints and IDE support
3. Removed unnecessary try-catch for yaml import (already in airflow-core dependencies)
4. All tests passing (19 config loader tests + 4 Param integration tests)
The refactoring follows the DRY principle and makes the code much more maintainable. Let me know if you need any other changes!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add dynamic parameter config source support for DAG params [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #59291:
URL: https://github.com/apache/airflow/pull/59291#discussion_r2610501795
##########
task-sdk/src/airflow/sdk/definitions/param.py:
##########
@@ -64,13 +81,67 @@ def __init__(
self.description = description
self.schema = kwargs.pop("schema") if "schema" in kwargs else kwargs
self.source = source
+ self.config_source = config_source
+
+ # If config_source is provided, load options and add to schema as enum
+ if config_source:
+ try:
+ options = self._load_options_from_config(config_source)
+ if options:
+ self.schema["enum"] = options
+ except Exception as e:
+ logger.warning(
+ "Failed to load options from config_source: %s. "
+ "Parameter will be treated as a regular string field.",
+ e,
+ )
def __copy__(self) -> Param:
return Param(
self.value,
self.description,
schema=self.schema,
source=self.source,
+ config_source=self.config_source,
+ )
+
+ def _load_options_from_config(self, config_source: dict[str, Any]) -> list[str]:
+ """
+ Load dropdown options from an external configuration file.
+
+ :param config_source: Dictionary containing configuration for loading options.
+ Must include 'file' key with path to config file.
+ Optional keys: 'filter' (dict), 'key_field' (str)
+ :return: List of option values loaded from the configuration file
+ :raises FileNotFoundError: If the configuration file is not found
+ :raises ValueError: If config_source is invalid or file format is unsupported
+ """
+ # Import here to avoid circular dependencies
+ from airflow.sdk.definitions.param_config_loader import (
+ load_options_from_ini,
+ load_options_from_json,
+ load_options_from_yaml,
+ )
+
+ file_path = config_source.get("file")
+ if not file_path:
+ raise ValueError("config_source must include 'file' key with path to configuration file")
+
+ filter_conditions = config_source.get("filter")
+ key_field = config_source.get("key_field")
+
+ # Detect file type and use appropriate loader
+ if file_path.endswith(".ini"):
+ key_field = key_field or "section"
+ return load_options_from_ini(file_path, filter_conditions, key_field)
+ if file_path.endswith(".json"):
+ key_field = key_field or "name"
+ return load_options_from_json(file_path, filter_conditions, key_field)
+ if file_path.endswith((".yaml", ".yml")):
+ key_field = key_field or "name"
+ return load_options_from_yaml(file_path, filter_conditions, key_field)
+ raise ValueError(
+ f"Unsupported config file type: {file_path}. Supported types are: .ini, .json, .yaml, .yml"
Review Comment:
Then we could just pass the file extension to `filter_conditions`.
##########
task-sdk/src/airflow/sdk/definitions/param.py:
##########
@@ -56,6 +72,7 @@ def __init__(
default: Any = NOTSET,
description: str | None = None,
source: Literal["dag", "task"] | None = None,
+ config_source: dict[str, Any] | None = None,
Review Comment:
How about adding a `TypeDict` for `config_source`? As `dict[str, Any]` seems to be too ambiguous for user.
##########
task-sdk/src/airflow/sdk/definitions/param.py:
##########
@@ -64,13 +81,67 @@ def __init__(
self.description = description
self.schema = kwargs.pop("schema") if "schema" in kwargs else kwargs
self.source = source
+ self.config_source = config_source
+
+ # If config_source is provided, load options and add to schema as enum
+ if config_source:
+ try:
+ options = self._load_options_from_config(config_source)
+ if options:
+ self.schema["enum"] = options
+ except Exception as e:
Review Comment:
We might need to add a new exception in `task-sdk/src/airflow/sdk/exceptions.py` instead of raising built-in `Exception`.
##########
task-sdk/src/airflow/sdk/definitions/param_config_loader.py:
##########
@@ -0,0 +1,243 @@
+# 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 loading DAG parameter options from external configuration files."""
+
+from __future__ import annotations
+
+import configparser
+import json
+from pathlib import Path
+from typing import Any
+
+
+def load_options_from_ini(
+ file_path: str | Path,
+ filter_conditions: dict[str, Any] | None = None,
+ key_field: str = "section",
+) -> list[str]:
+ """
+ Load dropdown options from an INI configuration file.
+
+ This function reads an INI file and extracts option values based on the specified
+ parameters. It can filter sections based on key-value pairs and choose which field
+ to use as the option value.
+
+ :param file_path: Path to the INI configuration file
+ :param filter_conditions: Optional dictionary of key-value pairs to filter sections.
+ Only sections where all specified keys match the given values will be included.
+ Example: {"TYPE": "Script"} will only include sections where TYPE=Script
+ :param key_field: Which field to use as the option value. Special value "section"
+ uses the section name itself. Otherwise, uses the value of the specified key
+ within each section.
+ :return: List of option values extracted from the INI file
+ :raises FileNotFoundError: If the specified file does not exist
+ :raises ValueError: If the INI file is malformed
+
+ Example INI file (interfaces.ini)::
+
+ [InterfaceA]
+ TYPE = Script
+ DESCRIPTION = Script interface A
+
+ [InterfaceB]
+ TYPE = EBICS
+ DESCRIPTION = Electronic banking interface
+
+ Example usage::
+
+ # Get all section names where TYPE=Script
+ options = load_options_from_ini(
+ "config/interfaces.ini", filter_conditions={"TYPE": "Script"}, key_field="section"
+ )
+ # Returns: ["InterfaceA"]
+
+ # Get all DESCRIPTION values
+ options = load_options_from_ini("config/interfaces.ini", key_field="DESCRIPTION")
+ # Returns: ["Script interface A", "Electronic banking interface"]
+ """
+ file_path = Path(file_path)
+ if not file_path.exists():
+ raise FileNotFoundError(f"Configuration file not found: {file_path}")
+
+ config = configparser.ConfigParser()
+ try:
+ config.read(file_path)
+ except configparser.Error as e:
+ raise ValueError(f"Invalid INI file format: {e}") from e
+
+ options = []
+ for section in config.sections():
+ # Check if section matches filter conditions
+ if filter_conditions:
+ matches = all(
+ config.get(section, key, fallback=None) == value for key, value in filter_conditions.items()
+ )
+ if not matches:
+ continue
+
+ # Get the value to use as option
+ if key_field == "section":
+ options.append(section)
+ else:
+ value = config.get(section, key_field, fallback=None)
+ if value:
+ options.append(value)
+
+ return options
+
+
+def load_options_from_json(
+ file_path: str | Path,
+ filter_conditions: dict[str, Any] | None = None,
+ key_field: str = "name",
+) -> list[str]:
+ """
+ Load dropdown options from a JSON configuration file.
+
+ This function reads a JSON file containing an array of objects and extracts
+ option values based on the specified parameters. It can filter objects based
+ on key-value pairs and choose which field to use as the option value.
+
+ :param file_path: Path to the JSON configuration file
+ :param filter_conditions: Optional dictionary of key-value pairs to filter objects.
+ Only objects where all specified keys match the given values will be included.
+ :param key_field: Which field to use as the option value from each object
+ :return: List of option values extracted from the JSON file
+ :raises FileNotFoundError: If the specified file does not exist
+ :raises ValueError: If the JSON file is malformed or not an array
+ :raises json.JSONDecodeError: If the file contains invalid JSON
+
+ Example JSON file (interfaces.json)::
+
+ [
+ {"name": "InterfaceA", "type": "Script", "description": "Script interface A"},
+ {"name": "InterfaceB", "type": "EBICS", "description": "Electronic banking"},
+ ]
+
+ Example usage::
+
+ # Get all names where type=Script
+ options = load_options_from_json(
+ "config/interfaces.json", filter_conditions={"type": "Script"}, key_field="name"
+ )
+ # Returns: ["InterfaceA"]
+ """
+ file_path = Path(file_path)
+ if not file_path.exists():
+ raise FileNotFoundError(f"Configuration file not found: {file_path}")
+
+ try:
+ with open(file_path) as f:
+ data = json.load(f)
+ except json.JSONDecodeError as e:
+ raise json.JSONDecodeError(f"Invalid JSON file format: {e.msg}", e.doc, e.pos) from e
+
+ if not isinstance(data, list):
+ raise ValueError("JSON file must contain an array of objects at the root level")
+
+ options = []
+ for item in data:
+ if not isinstance(item, dict):
+ continue
+
+ # Check if item matches filter conditions
+ if filter_conditions:
+ matches = all(item.get(key) == value for key, value in filter_conditions.items())
+ if not matches:
+ continue
+
+ # Get the value to use as option
+ value = item.get(key_field)
+ if value is not None:
+ options.append(str(value))
+
+ return options
+
+
+def load_options_from_yaml(
+ file_path: str | Path,
+ filter_conditions: dict[str, Any] | None = None,
+ key_field: str = "name",
+) -> list[str]:
+ """
+ Load dropdown options from a YAML configuration file.
+
+ This function reads a YAML file containing an array of objects and extracts
+ option values based on the specified parameters. Similar to JSON loading but
+ supports YAML format.
+
+ :param file_path: Path to the YAML configuration file
+ :param filter_conditions: Optional dictionary of key-value pairs to filter objects
+ :param key_field: Which field to use as the option value from each object
+ :return: List of option values extracted from the YAML file
+ :raises FileNotFoundError: If the specified file does not exist
+ :raises ValueError: If the YAML file is malformed or not an array
+ :raises ImportError: If PyYAML is not installed
+
+ Example YAML file (interfaces.yaml)::
+
+ - name: InterfaceA
+ type: Script
+ description: Script interface A
+ - name: InterfaceB
+ type: EBICS
+ description: Electronic banking
+
+ Example usage::
+
+ # Get all names where type=Script
+ options = load_options_from_yaml(
+ "config/interfaces.yaml", filter_conditions={"type": "Script"}, key_field="name"
+ )
+ # Returns: ["InterfaceA"]
+ """
+ try:
+ import yaml
+ except ImportError as e:
+ raise ImportError("PyYAML is required to load YAML files. Install it with: pip install PyYAML") from e
Review Comment:
It seems we don't need to have the try-catch for importing `yaml` package as task-sdk is depends on airflow-core
https://github.com/apache/airflow/blob/7c6fcecc535de447482b18c85fb98235655e745d/task-sdk/pyproject.toml#L52-L53
and `pyyaml` is already included in airflow-core.
https://github.com/apache/airflow/blob/7c6fcecc535de447482b18c85fb98235655e745d/airflow-core/pyproject.toml#L157-L158
##########
task-sdk/src/airflow/sdk/definitions/param_config_loader.py:
##########
@@ -0,0 +1,243 @@
+# 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 loading DAG parameter options from external configuration files."""
+
+from __future__ import annotations
+
+import configparser
+import json
+from pathlib import Path
+from typing import Any
+
+
+def load_options_from_ini(
+ file_path: str | Path,
+ filter_conditions: dict[str, Any] | None = None,
+ key_field: str = "section",
+) -> list[str]:
+ """
+ Load dropdown options from an INI configuration file.
+
+ This function reads an INI file and extracts option values based on the specified
+ parameters. It can filter sections based on key-value pairs and choose which field
+ to use as the option value.
+
+ :param file_path: Path to the INI configuration file
+ :param filter_conditions: Optional dictionary of key-value pairs to filter sections.
+ Only sections where all specified keys match the given values will be included.
+ Example: {"TYPE": "Script"} will only include sections where TYPE=Script
+ :param key_field: Which field to use as the option value. Special value "section"
+ uses the section name itself. Otherwise, uses the value of the specified key
+ within each section.
+ :return: List of option values extracted from the INI file
+ :raises FileNotFoundError: If the specified file does not exist
+ :raises ValueError: If the INI file is malformed
+
+ Example INI file (interfaces.ini)::
+
+ [InterfaceA]
+ TYPE = Script
+ DESCRIPTION = Script interface A
+
+ [InterfaceB]
+ TYPE = EBICS
+ DESCRIPTION = Electronic banking interface
+
+ Example usage::
+
+ # Get all section names where TYPE=Script
+ options = load_options_from_ini(
+ "config/interfaces.ini", filter_conditions={"TYPE": "Script"}, key_field="section"
+ )
+ # Returns: ["InterfaceA"]
+
+ # Get all DESCRIPTION values
+ options = load_options_from_ini("config/interfaces.ini", key_field="DESCRIPTION")
+ # Returns: ["Script interface A", "Electronic banking interface"]
+ """
+ file_path = Path(file_path)
+ if not file_path.exists():
+ raise FileNotFoundError(f"Configuration file not found: {file_path}")
Review Comment:
Would it be better to consolidate the three method as `load_options_from_file` and provide `extension: Literal["ini", "json", "yaml"]` parameter?
Since the validation of file path and filter of `filter_conditions` and `key_field` are duplicated.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add dynamic parameter config source support for DAG params [airflow]
Posted by "chen0427ok (via GitHub)" <gi...@apache.org>.
chen0427ok commented on code in PR #59291:
URL: https://github.com/apache/airflow/pull/59291#discussion_r2611568124
##########
task-sdk/src/airflow/sdk/definitions/param.py:
##########
@@ -56,6 +72,7 @@ def __init__(
default: Any = NOTSET,
description: str | None = None,
source: Literal["dag", "task"] | None = None,
+ config_source: dict[str, Any] | None = None,
Review Comment:
That's a great suggestion. Using TypedDict will definitely make the expected structure much clearer for users and remove the ambiguity of dict[str, Any]. I'll define a ConfigSource TypedDict and apply it.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add dynamic parameter config source support for DAG params [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #59291:
URL: https://github.com/apache/airflow/pull/59291#issuecomment-3640353228
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:
- Pay attention to the quality of your code (ruff, mypy and type annotations). Our [prek-hooks]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-prek-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/airflow-core/docs/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add dynamic parameter config source support for DAG params [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #59291:
URL: https://github.com/apache/airflow/pull/59291#issuecomment-3844522358
This pull request has been automatically marked as stale because the author has not responded to a request for more information. It will be closed in 7 days if no further activity occurs. Thank you for your contributions.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add dynamic parameter config source support for DAG params [airflow]
Posted by "chen0427ok (via GitHub)" <gi...@apache.org>.
chen0427ok commented on PR #59291:
URL: https://github.com/apache/airflow/pull/59291#issuecomment-3641072774
> While the idea is sound - we are in the middle of making task-sdk independent from airflow-core. We should not add anything in task-sdk that uses anything from `airflow-core` - so please move the code entirely to task-sdk.
Thanks for the feedback! I've refactored the code to move param_config_loader entirely into the task-sdk as requested.
Summary of changes:
Moved files: Relocated source and test files from airflow-core to their respective paths in task-sdk.
Updated imports: Changed references in param.py to use airflow.sdk.definitions.param_config_loader.
Cleanup: Removed the try-except ImportError block since the dependency is now internal to the SDK.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add dynamic parameter config source support for DAG params [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #59291: feat: Add dynamic parameter config source support for DAG params
URL: https://github.com/apache/airflow/pull/59291
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add dynamic parameter config source support for DAG params [airflow]
Posted by "chen0427ok (via GitHub)" <gi...@apache.org>.
chen0427ok commented on code in PR #59291:
URL: https://github.com/apache/airflow/pull/59291#discussion_r2611565846
##########
task-sdk/src/airflow/sdk/definitions/param_config_loader.py:
##########
@@ -0,0 +1,243 @@
+# 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 loading DAG parameter options from external configuration files."""
+
+from __future__ import annotations
+
+import configparser
+import json
+from pathlib import Path
+from typing import Any
+
+
+def load_options_from_ini(
+ file_path: str | Path,
+ filter_conditions: dict[str, Any] | None = None,
+ key_field: str = "section",
+) -> list[str]:
+ """
+ Load dropdown options from an INI configuration file.
+
+ This function reads an INI file and extracts option values based on the specified
+ parameters. It can filter sections based on key-value pairs and choose which field
+ to use as the option value.
+
+ :param file_path: Path to the INI configuration file
+ :param filter_conditions: Optional dictionary of key-value pairs to filter sections.
+ Only sections where all specified keys match the given values will be included.
+ Example: {"TYPE": "Script"} will only include sections where TYPE=Script
+ :param key_field: Which field to use as the option value. Special value "section"
+ uses the section name itself. Otherwise, uses the value of the specified key
+ within each section.
+ :return: List of option values extracted from the INI file
+ :raises FileNotFoundError: If the specified file does not exist
+ :raises ValueError: If the INI file is malformed
+
+ Example INI file (interfaces.ini)::
+
+ [InterfaceA]
+ TYPE = Script
+ DESCRIPTION = Script interface A
+
+ [InterfaceB]
+ TYPE = EBICS
+ DESCRIPTION = Electronic banking interface
+
+ Example usage::
+
+ # Get all section names where TYPE=Script
+ options = load_options_from_ini(
+ "config/interfaces.ini", filter_conditions={"TYPE": "Script"}, key_field="section"
+ )
+ # Returns: ["InterfaceA"]
+
+ # Get all DESCRIPTION values
+ options = load_options_from_ini("config/interfaces.ini", key_field="DESCRIPTION")
+ # Returns: ["Script interface A", "Electronic banking interface"]
+ """
+ file_path = Path(file_path)
+ if not file_path.exists():
+ raise FileNotFoundError(f"Configuration file not found: {file_path}")
Review Comment:
Thanks for the review and the great suggestion! I agree that consolidating them into a single load_options_from_file function will reduce code duplication and make it easier to maintain. I'll refactor the code to merge the validation and filtering logic.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add dynamic parameter config source support for DAG params [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #59291:
URL: https://github.com/apache/airflow/pull/59291#issuecomment-3802428279
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add dynamic parameter config source support for DAG params [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #59291:
URL: https://github.com/apache/airflow/pull/59291#issuecomment-3640796060
While the idea is sound - we are in the middle of making task-sdk independent from airflow-core. We should not add anything in task-sdk that uses anything from `airflow-core` - so please move the code entirely to task-sdk.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add dynamic parameter config source support for DAG params [airflow]
Posted by "chen0427ok (via GitHub)" <gi...@apache.org>.
chen0427ok commented on code in PR #59291:
URL: https://github.com/apache/airflow/pull/59291#discussion_r2611567125
##########
task-sdk/src/airflow/sdk/definitions/param_config_loader.py:
##########
@@ -0,0 +1,243 @@
+# 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 loading DAG parameter options from external configuration files."""
+
+from __future__ import annotations
+
+import configparser
+import json
+from pathlib import Path
+from typing import Any
+
+
+def load_options_from_ini(
+ file_path: str | Path,
+ filter_conditions: dict[str, Any] | None = None,
+ key_field: str = "section",
+) -> list[str]:
+ """
+ Load dropdown options from an INI configuration file.
+
+ This function reads an INI file and extracts option values based on the specified
+ parameters. It can filter sections based on key-value pairs and choose which field
+ to use as the option value.
+
+ :param file_path: Path to the INI configuration file
+ :param filter_conditions: Optional dictionary of key-value pairs to filter sections.
+ Only sections where all specified keys match the given values will be included.
+ Example: {"TYPE": "Script"} will only include sections where TYPE=Script
+ :param key_field: Which field to use as the option value. Special value "section"
+ uses the section name itself. Otherwise, uses the value of the specified key
+ within each section.
+ :return: List of option values extracted from the INI file
+ :raises FileNotFoundError: If the specified file does not exist
+ :raises ValueError: If the INI file is malformed
+
+ Example INI file (interfaces.ini)::
+
+ [InterfaceA]
+ TYPE = Script
+ DESCRIPTION = Script interface A
+
+ [InterfaceB]
+ TYPE = EBICS
+ DESCRIPTION = Electronic banking interface
+
+ Example usage::
+
+ # Get all section names where TYPE=Script
+ options = load_options_from_ini(
+ "config/interfaces.ini", filter_conditions={"TYPE": "Script"}, key_field="section"
+ )
+ # Returns: ["InterfaceA"]
+
+ # Get all DESCRIPTION values
+ options = load_options_from_ini("config/interfaces.ini", key_field="DESCRIPTION")
+ # Returns: ["Script interface A", "Electronic banking interface"]
+ """
+ file_path = Path(file_path)
+ if not file_path.exists():
+ raise FileNotFoundError(f"Configuration file not found: {file_path}")
+
+ config = configparser.ConfigParser()
+ try:
+ config.read(file_path)
+ except configparser.Error as e:
+ raise ValueError(f"Invalid INI file format: {e}") from e
+
+ options = []
+ for section in config.sections():
+ # Check if section matches filter conditions
+ if filter_conditions:
+ matches = all(
+ config.get(section, key, fallback=None) == value for key, value in filter_conditions.items()
+ )
+ if not matches:
+ continue
+
+ # Get the value to use as option
+ if key_field == "section":
+ options.append(section)
+ else:
+ value = config.get(section, key_field, fallback=None)
+ if value:
+ options.append(value)
+
+ return options
+
+
+def load_options_from_json(
+ file_path: str | Path,
+ filter_conditions: dict[str, Any] | None = None,
+ key_field: str = "name",
+) -> list[str]:
+ """
+ Load dropdown options from a JSON configuration file.
+
+ This function reads a JSON file containing an array of objects and extracts
+ option values based on the specified parameters. It can filter objects based
+ on key-value pairs and choose which field to use as the option value.
+
+ :param file_path: Path to the JSON configuration file
+ :param filter_conditions: Optional dictionary of key-value pairs to filter objects.
+ Only objects where all specified keys match the given values will be included.
+ :param key_field: Which field to use as the option value from each object
+ :return: List of option values extracted from the JSON file
+ :raises FileNotFoundError: If the specified file does not exist
+ :raises ValueError: If the JSON file is malformed or not an array
+ :raises json.JSONDecodeError: If the file contains invalid JSON
+
+ Example JSON file (interfaces.json)::
+
+ [
+ {"name": "InterfaceA", "type": "Script", "description": "Script interface A"},
+ {"name": "InterfaceB", "type": "EBICS", "description": "Electronic banking"},
+ ]
+
+ Example usage::
+
+ # Get all names where type=Script
+ options = load_options_from_json(
+ "config/interfaces.json", filter_conditions={"type": "Script"}, key_field="name"
+ )
+ # Returns: ["InterfaceA"]
+ """
+ file_path = Path(file_path)
+ if not file_path.exists():
+ raise FileNotFoundError(f"Configuration file not found: {file_path}")
+
+ try:
+ with open(file_path) as f:
+ data = json.load(f)
+ except json.JSONDecodeError as e:
+ raise json.JSONDecodeError(f"Invalid JSON file format: {e.msg}", e.doc, e.pos) from e
+
+ if not isinstance(data, list):
+ raise ValueError("JSON file must contain an array of objects at the root level")
+
+ options = []
+ for item in data:
+ if not isinstance(item, dict):
+ continue
+
+ # Check if item matches filter conditions
+ if filter_conditions:
+ matches = all(item.get(key) == value for key, value in filter_conditions.items())
+ if not matches:
+ continue
+
+ # Get the value to use as option
+ value = item.get(key_field)
+ if value is not None:
+ options.append(str(value))
+
+ return options
+
+
+def load_options_from_yaml(
+ file_path: str | Path,
+ filter_conditions: dict[str, Any] | None = None,
+ key_field: str = "name",
+) -> list[str]:
+ """
+ Load dropdown options from a YAML configuration file.
+
+ This function reads a YAML file containing an array of objects and extracts
+ option values based on the specified parameters. Similar to JSON loading but
+ supports YAML format.
+
+ :param file_path: Path to the YAML configuration file
+ :param filter_conditions: Optional dictionary of key-value pairs to filter objects
+ :param key_field: Which field to use as the option value from each object
+ :return: List of option values extracted from the YAML file
+ :raises FileNotFoundError: If the specified file does not exist
+ :raises ValueError: If the YAML file is malformed or not an array
+ :raises ImportError: If PyYAML is not installed
+
+ Example YAML file (interfaces.yaml)::
+
+ - name: InterfaceA
+ type: Script
+ description: Script interface A
+ - name: InterfaceB
+ type: EBICS
+ description: Electronic banking
+
+ Example usage::
+
+ # Get all names where type=Script
+ options = load_options_from_yaml(
+ "config/interfaces.yaml", filter_conditions={"type": "Script"}, key_field="name"
+ )
+ # Returns: ["InterfaceA"]
+ """
+ try:
+ import yaml
+ except ImportError as e:
+ raise ImportError("PyYAML is required to load YAML files. Install it with: pip install PyYAML") from e
Review Comment:
Thanks for pointing that out! I didn't realize pyyaml was already a dependency in airflow-core. I'll remove the try-catch block and import yaml directly.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org