dinigo commented on issue #20857:
URL: https://github.com/apache/airflow/issues/20857#issuecomment-1012622806
From the docs
> ```python
> class HelloOperator(BaseOperator):
>
> template_fields = ["guest_name"]
> template_ext = [".sql"]
>
> def __init__(self, name: str, **kwargs) -> None:
> super().__init__(**kwargs)
> self.guest_name = name
> ```
> _In the example, the template_fields should be `['guest_name']` and not
`['name']`_
I can see how what's templated is the "property" rather than the
"constructor argument". The docs make it semi-clear. My error came from the
`output` collyding with an internal `@property` of the `BaseOperator`.
Here's a test to prove docs are right
```python
import uuid
from datetime import datetime, timedelta, timezone
from typing import Any
import pytest
from airflow import DAG
from airflow.models import BaseOperator
from airflow.utils.state import DagRunState, State
from airflow.utils.types import DagRunType
class TemplateOutputOperator(BaseOperator):
template_fields = ['_output']
def __init__(self, output: str = None, *args, **kwargs):
self._output = output
super(TemplateOutputOperator, self).__init__(*args, **kwargs)
def execute(self, context: Any):
print(f"{self._output=}")
DATA_INTERVAL_START = datetime(2021, 9, 13, tzinfo=timezone.utc)
DATA_INTERVAL_END = DATA_INTERVAL_START + timedelta(days=1)
TEST_DAG_ID = "test_template_output_dag"
TEST_TASK_ID = "test_template_output_task"
@pytest.fixture()
def dag():
with DAG(
dag_id='test_template_output_dag',
schedule_interval="@daily",
default_args={"start_date": DATA_INTERVAL_START},
) as dag:
TemplateOutputOperator(
task_id='test_template_output_task',
output='{{ task_instance_key_str }}'
)
return dag
def test_operator_templates_output_param(dag):
dag_run = dag.create_dagrun(
state=DagRunState.RUNNING,
execution_date=datetime.now(timezone.utc),
data_interval=(DATA_INTERVAL_START, DATA_INTERVAL_END),
start_date=DATA_INTERVAL_END,
run_type=DagRunType.MANUAL,
run_id=uuid.uuid4().hex
)
ti = dag_run.get_task_instance(task_id=TEST_TASK_ID)
ti.task = dag.get_task(task_id=TEST_TASK_ID)
ti.run(ignore_ti_state=True)
assert str(ti.task._output).startswith(TEST_DAG_ID + '__' + TEST_TASK_ID)
```
--
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]
