A custom Airflow operator puts its task work in `execute(self, context)`. Airflow calls this method when the operator runs as part of a DAG run. `context` is a dictionary of Airflow context variables, including dates that the work may need. section 9.3.1
For a ratings pipeline, an operator can accept a connection ID, date range, and output path at construction. Its `execute` method then uses a hook to fetch ratings and writes them to a JSON file. Instance attributes supply the configured values; `self.log` supplies the operator logger. Close the hook in a `finally` block so an API failure does not leave its session open. section 9.3.2
Fixed dates make that operator fetch the same period on every run. Declaring its date and output-path attributes as `template_fields` lets their values refer to `data_interval_start` and `data_interval_end` instead. section 9.3.2
Listing 9.24 Adding template fields
class MovielensFetchRatingsOperator(BaseOperator):
...
template_fields = (
"_start_date",#1
"_end_date", #1
"_output_path" #1
)
...
def __init__(
self,
conn_id:str,
output_path:str,
start_date:str="{{data_interval_start | ds}}",
end_date:str="{{data_interval_end | ds}}",
**kwargs,
):
super(MovielensFetchRatingsOperator, self).__init__(**kwargs)
self._conn_id = conn_id
self._output_path = output_path
self._start_date = start_date
self._end_date = end_date
The template-field declaration shows which configured attributes can vary with the data interval before execution. section 9.3.2
For a focused unit test, instantiate an operator and call `execute(context={})` directly, then assert its result. An empty context works only when the tested code does not need context values; otherwise supply the required keys. This call does not reproduce a live Airflow run’s surrounding setup, including template rendering. Mock external connections when testing an operator in isolation. section 10.1.4 section 10.5
Listing 10.11 Example test function testing the BashOperator
def test_example():
task = BashOperator(
task_id="test",
bash_command="echo 'hello!'",
)
result = task.execute(context={})
assert result == "hello!"
A small BashOperator test shows the direct call and result assertion without requiring task-instance context. section 10.1.4
