jroachgolf84 commented on issue #62063: URL: https://github.com/apache/airflow/issues/62063#issuecomment-5504275406
### Status update: `airflow.utils` usage in providers Full inventory below of every `airflow.utils` import found in `providers/*/src`, grouped by module rather than by provider, with what has already been resolved, what still needs work and has a Task SDK equivalent, and what has no Task SDK equivalent at all. #### Already resolved These already prefer the Task SDK path, with `airflow.utils` kept only as a fallback for pre-3.0 Airflow, or have no remaining production usage: - `log.secrets_masker` (`mask_secret`, `SecretsMasker`, `_secrets_masker`, `Redactable`, `Redacted`): amazon, common.ai, openlineage already prefer `airflow.sdk.log` / `airflow.sdk._shared.secrets_masker`. - `context.context_get_outlet_events`: standard already version gates to `airflow.sdk.execution_time.context`. - `dag_parsing_context._airflow_parsing_context_manager`: celery already prefers `airflow.sdk.definitions._internal.dag_parsing_context`. - `trigger_rule.TriggerRule`: only referenced in edge3 example dags, already version gated to `airflow.sdk.TriggerRule`. - `module_loading` (`import_string`, `iter_namespace`, `qualname`): common.compat and elasticsearch already handle this correctly (the module was removed from airflow-core entirely in December 2025, replaced by a shared distribution). - `task_group.TaskGroup`, `edgemodifier.Label`: no remaining production usage, only found in provider test files. - `state.TaskInstanceState` / `DagRunState` for edge3, opensearch, common.ai, informatica: covered by #72408, #72409, #72410, #72411. #### Outstanding, Task SDK equivalent exists **`state.TaskInstanceState` / `DagRunState`, remaining providers** Same replacement as the four issues above (`airflow.sdk.TaskInstanceState` / `DagRunState`), but these providers still support `apache-airflow>=2.11.0`, so each needs a version guard rather than a bare swap. Most already import an `AIRFLOW_V_3_0_PLUS` style flag, so the guard is a small addition: - amazon: `aws/sensors/mwaa.py`, `aws/triggers/emr.py`, `aws/triggers/mwaa.py` - google: `cloud/triggers/dataproc.py`, `cloud/triggers/bigquery.py`, `cloud/sensors/cloud_composer.py` - cncf.kubernetes: `triggers/pod.py`, `executors/kubernetes_executor.py`, `executors/kubernetes_executor_utils.py`, `executors/kubernetes_executor_types.py` (already `TYPE_CHECKING` only, safe as a bare swap) - databricks: `plugins/databricks_workflow.py` - standard: `sensors/external_task.py`, `operators/trigger_dagrun.py`, `triggers/external_task.py` (already `TYPE_CHECKING` only, safe as a bare swap) - openlineage: `plugins/adapter.py` (already `TYPE_CHECKING` only), `plugins/listener.py`, `api/datasets.py`, `utils/utils.py` (already `TYPE_CHECKING` only), `extractors/manager.py` - celery: `executors/celery_executor.py` - elasticsearch: `log/es_task_handler.py` - apache.kafka: `plugins/event_producer.py` (already `TYPE_CHECKING` only, safe as a bare swap) **`operator_helpers.determine_kwargs`** `airflow.utils.operator_helpers.determine_kwargs` is already a deprecated re-export pointing at `airflow.sdk.bases.decorator.determine_kwargs`, so the swap itself is low risk. Still supports Airflow 2.11+, so needs the same version guard treatment: - standard: `sensors/python.py` - http: `sensors/http.py`, `operators/http.py` **`types.NOTSET` / `ArgNotSet`** Replacement is `airflow.sdk.definitions._internal.types.NOTSET` / `ArgNotSet`, already the established pattern in amazon's `version_compat.py`. Not yet applied in: - google: `cloud/operators/bigquery.py`, `cloud/hooks/bigquery.py`, `cloud/hooks/compute_ssh.py`, `cloud/log/stackdriver_task_handler.py` - cncf.kubernetes: `cli/kubernetes_command.py` - standard: `operators/trigger_dagrun.py`, `operators/latest_only.py` - common.sql: `decorators/sql.py`, `decorators/analytics.py` - ssh: `operators/ssh.py`, `hooks/ssh.py` - openlineage: `utils/utils.py` - slack: `utils/__init__.py` **`operator_helpers.ExecutionCallableRunner`** Task SDK equivalent is `airflow.sdk.execution_time.callback_runner.create_executable_runner`, but it is a function returning a callable rather than a class you instantiate, so this is not a plain import swap. Standard's `operators/python.py` already handles this correctly for Airflow 3.0+ and only falls back to the old class for pre-3.0 installs, so no further action is needed there. No other provider uses this one directly. **`log.logging_mixin.LoggingMixin`** Task SDK has its own `airflow.sdk.definitions._internal.logging_mixin.LoggingMixin`, used internally throughout task-sdk itself, but it lives under a `_internal` path with no public API guarantee, and no provider currently imports it this way. Roughly 55 files across most providers still import `LoggingMixin` from `airflow.utils.log.logging_mixin`. This needs a decision on whether `_internal` is an acceptable import target before treating it as a swap candidate, rather than being ready to go like the items above. #### No Task SDK equivalent, out of scope for a direct swap These would need to be inlined as private provider code instead (option 1 in the issue description), a separate effort from the import swaps above: - `helpers`: `chunks`, `exactly_one`, `prune_dict`, `merge_dicts`, `convert_camel_to_snake`, `as_flattened_list`, `validate_key`, `build_airflow_dagrun_url`, `build_airflow_url_with_query` - `strings`: `get_random_string`, `to_boolean` - `hashlib_wrapper.md5` (task-sdk just calls `hashlib.md5` directly, no wrapper to import) - `platform`: `IS_WINDOWS`, `get_airflow_git_version` (`getuser` has an internal analog in the Task SDK's execution API client, but it is not meant as a general utility import) - `net.getfqdn` - `cli`: `get_bagged_dag`, `setup_locations`, `suppress_logs_and_warning` - `db`, `db_manager`, `session` (`create_session`, `provide_session`, `NEW_SESSION`), `sqlalchemy.UtcDateTime`, `task_instance_session`: no equivalent by design, workers and tasks do not access the database directly - `log.file_task_handler`, `log.json_formatter.JSONFormatter`, `log.log_reader.TaskLogReader`, `log.trigger_handler.ctx_indiv_trigger`, `serve_logs`: server and scheduler side log serving components - `log.logging_mixin`: `ExternalLoggingMixin`, `remove_escape_codes`, `set_context` - `deprecation_tools`: `add_deprecated_classes`, `DeprecatedImportWarning` - `providers_configuration_loader.providers_configuration_loaded` - `process_utils`: `patch_environ`, `set_component_mp_start_method` - `context.AirflowContextDeprecationWarning` (pure Airflow 2.x compat, already correctly guarded where used) - `email`: `build_mime_message`, `get_email_address_list` - `file`: `correct_maybe_zipped`, `get_unique_dag_module_name` - `json`: `XComDecoder`, `XComEncoder` - `state`: `State`, `JobState`, `TerminalTIState` (static helper and scheduler only, no equivalent, distinct from `TaskInstanceState` / `DagRunState` above) - `types`: `DagRunType`, `DagRunTriggeredByType` (exist internally in the Task SDK's generated API datamodels but are not exposed as stable public API) --- Drafted-by: Claude Code (Sonnet 5) (no human review before posting) -- 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]
