Fix template file lookup for Dags packaged in zip files

[sjyangkevin]

Why

Dags packaged as zip archives could not use template files shipped inside the archive. For a zipped Dag, dag.folder points into the archive (for example .../dags.zip), and Jinja's FileSystemLoader only reads real directories. A templated field such as "query": "test_sql/test.sql" therefore failed every run with:

jinja2.exceptions.TemplateNotFound: 'test_sql/test.sql' not found in search path: '.../test.zip'

At parse time the same lookup failed silently (only logged), and pointing template_searchpath into the archive failed the same way.

create_template_env() now reads search-path entries that are, or point inside, a zip archive directly from the archive. Normal directories keep exactly the same FileSystemLoader as before. Both template environment builders (DAG.get_template_env() and the operator-level render_template_as_native_obj override) go through this function.

closes: #59310
related: #60999

How zip entries are handled

• The archive is found by walking up to the nearest existing path and checking it with zipfile.is_zipfile(), so detection does not depend on the .zip extension or its case.
• Members are read in memory by exact name; nothing is extracted to disk.
• Template names are validated with Jinja's split_template_path, as in FileSystemLoader, so they cannot leave the search root. Search-path prefixes containing .. are not treated as archive paths.
• Search-path order is preserved across zip and directory entries, and a missing template still raises TemplateNotFound with the not found in search path(s): ... message.
• An unreadable archive raises instead of falling through to a later search path.
• The archive is opened per lookup, so no file handle is shared across the task supervisor's fork.

Testing

• 18 new test cases. 9 fail without this change: loading from a zip (including DAGS.ZIP), zip-first ordering, both template environment builders with a .sql path nested in a dict, and parse-time resolve_template_files() on a Dag imported by the zip importer. The other 9 pass on main by design: they pin unchanged behaviour for non-zip search paths, the not-found message, and containment of template names. Each was checked to fail when the corresponding check in the new loader is removed.
• End-to-end in Breeze, using airflow dags test and the UI from breeze start-airflow, with zipped Dags using a top-level template file, a .sql path nested in a dict, template_searchpath inside the archive, and {% include %}. All failed with TemplateNotFound before this change and succeed after it. Unzipped copies of the same Dags succeed both before and after.
• The zip attached to the issue now gets past template rendering and reaches BigQueryInsertJobOperator.execute(). The BigQuery call itself was not run (no GCP credentials), and this change does not touch it.
• prek pre-commit and manual stages pass, including mypy-task-sdk. From breeze verify: task-sdk-tests (3656 passed), core non-DB tests and core DB tests pass.

Before/after reproduction in Breeze (airflow dags test)

repro.zip contains repro_dags.py, scripts/hello.sh and sql/query.sql. The same files unzipped provide the dir_* control Dags. reporter.zip is the test.zip attached to the issue, unchanged.

Dag	Template usage
zip_top_level	bash_command="scripts/hello.sh"
zip_nested_dict	.sql path nested in a dict field, like BigQueryInsertJobOperator
zip_searchpath	template_searchpath pointing to a folder inside the archive
zip_include	{% include 'scripts/hello.sh' %}
test_import_path_dag	the Dag attached to the issue
dir_top_level	unzipped control

Before (main), trimmed to each Dag's result and its own task error:

===== import errors
No data found
===== parse-time resolve_template_files warnings
Failed to resolve template field 'bash_command' [airflow.sdk.definitions._internal.templater]
######## dir_top_level -> state=success
hello from dir_top_level / t
######## zip_top_level -> state=failed
jinja2.exceptions.TemplateNotFound: 'scripts/hello.sh' not found in search path: '/files/repro_59310/dags/repro.zip'
######## zip_nested_dict -> state=failed
jinja2.exceptions.TemplateNotFound: 'sql/query.sql' not found in search path: '/files/repro_59310/dags/repro.zip'
######## zip_searchpath -> state=failed
jinja2.exceptions.TemplateNotFound: 'hello.sh' not found in search paths: '/files/repro_59310/dags/repro.zip', '/files/repro_59310/dags/repro.zip/scripts'
######## zip_include -> state=failed
jinja2.exceptions.TemplateNotFound: 'scripts/hello.sh' not found in search path: '/files/repro_59310/dags/repro.zip'
######## test_import_path_dag -> state=failed
jinja2.exceptions.TemplateNotFound: 'test_sql/test.sql' not found in search path: '/files/repro_59310/dags/reporter.zip'

After (this PR):

===== import errors
No data found
===== parse-time resolve_template_files warnings
######## dir_top_level -> state=success
hello from dir_top_level / t
######## zip_top_level -> state=success
hello from zip_top_level / t
######## zip_nested_dict -> state=success
RENDERED: {'query': {'query': 'SELECT column_a FROM test -- zip_nested_dict'}}
######## zip_searchpath -> state=success
hello from zip_searchpath / t
######## zip_include -> state=success
hello from zip_include / t
######## test_import_path_dag -> state=failed

test_import_path_dag now gets through template rendering and fails in BigQueryInsertJobOperator.execute() with AirflowNotFoundException: The conn_id `google_cloud_default` isn't defined, since no GCP connection.

repro_dags.py

import os

from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import DAG, BaseOperator

# Same source is used zipped and unzipped; prefix keeps dag_ids unique in the bundle.
PREFIX = "zip" if ".zip" in __file__ else "dir"
HERE = os.path.dirname(__file__)


class NestedSqlOperator(BaseOperator):
    """Mimics BigQueryInsertJobOperator: a .sql path nested inside a dict template field."""

    template_fields = ("configuration",)
    template_ext = (".sql",)

    def __init__(self, configuration, **kwargs):
        super().__init__(**kwargs)
        self.configuration = configuration

    def execute(self, context):
        print("RENDERED:", self.configuration)
        return self.configuration["query"]["query"]


with DAG(dag_id=f"{PREFIX}_top_level", schedule=None):
    BashOperator(task_id="t", bash_command="scripts/hello.sh")

with DAG(dag_id=f"{PREFIX}_nested_dict", schedule=None):
    NestedSqlOperator(task_id="t", configuration={"query": {"query": "sql/query.sql"}})

with DAG(dag_id=f"{PREFIX}_searchpath", schedule=None, template_searchpath=[os.path.join(HERE, "scripts")]):
    BashOperator(task_id="t", bash_command="hello.sh")

with DAG(dag_id=f"{PREFIX}_include", schedule=None):
    BashOperator(task_id="t", bash_command="{% include 'scripts/hello.sh' %}")

scripts/hello.sh is echo "hello from {{ dag.dag_id }} / {{ ti.task_id }}" and sql/query.sql is SELECT column_a FROM test -- {{ dag.dag_id }}.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code (Opus 5.5) following the guidelines