My custom Airflow operator receives a list of file names as input. I want to:
- Render these files
- Save this list of file names to some operator's instance attribute
The problem is that I can't figure out how to keep the original file names - it seems that the list containing the file names gets rendered before instance variables assignment and file names get replaced by their rendered contents.
Here is the example code:
from typing import Sequence
from airflow import DAG
from airflow.models import BaseOperator
from pendulum import datetime
class RenderingOperator(BaseOperator):
template_fields: Sequence[str] = ("inputs_sql",)
template_ext: Sequence[str] = (".sql",)
def __init__(self, inputs_sql, **kwargs):
super().__init__(**kwargs)
self.inputs_sql_original = inputs_sql
self.inputs_sql = inputs_sql
def execute(self, context):
print("self.inputs_sql_original gets rendered:")
print(self.inputs_sql_original)
print("self.inputs_sql gets rendered:")
print(self.inputs_sql)
with DAG(
dag_id="check_rendering"
) as dag:
RenderingOperator(
task_id="render_sql",
inputs_sql=["query/query_0.sql"],
dag=dag,
)
The outputs printed:
self.inputs_sql_original gets rendered:
['select * from table\n']
self.inputs_sql gets rendered:
['select * from table\n']
Can you please suggest how can I keep both the original file names and rendered files?
PS If I use a string instead of a list for inputs_sql the original file names are preserved in self.inputs_sql_original.