Fix using .json template extension in GMP operators

[turbaszek]

Closes #9541

---

Make sure to mark the boxes below before creating PR: [x]

• Description above provides context of the change
• Unit tests coverage for changes (not needed for documentation changes)
• Target Github ISSUE in description if exists
• Commits follow "How to write a good git commit message"
• Relevant documentation is updated including usage instructions.
• I will engage committers as explained in Contribution Workflow Example.

---

In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.
Read the Pull Request Guidelines for more information.

[turbaszek]

@olchas @efolgar would you mind taking a look?

[efolgar]

This would work for templates passed in with a *.json extension, but doesn't cover the case where report is passed as a JSON string template. For example:

create_report = GoogleCampaignManagerInsertReportOperator(
        task_id="create_report",
        gcp_conn_id=self.config.get("gcp_connection_id"),
        profile_id=self.config.get("cm_profile_id"),
        report="""
            {
               'name': '{{ params.title }}',
               'accountId': '{{ params.account }}',
               'fileName': '{{ params.file_name }}',
               'type': 'STANDARD',
               'criteria': {
               ...""",
        params=report_params,
        dag=self.dag)

Safest approach may be to parse the json string within the execute method.

[turbaszek]

According to documentation user should provide a dictionary so I would say that passing str doesn't have to be supported.

[mik-laj]

You can create a task as below to create a task based on a JSON file from another source e.g. variable.

create_report = GoogleCampaignManagerInsertReportOperator(
        task_id="create_report",
        gcp_conn_id=self.config.get("gcp_connection_id"),
        profile_id=self.config.get("cm_profile_id"),
        report=json.loads("""
            {
               'name': '{{ params.title }}',
               'accountId': '{{ params.account }}',
               'fileName': '{{ params.file_name }}',
               'type': 'STANDARD',
               'criteria': {
               ..."""),
        params=report_params,
        dag=self.dag)

Otherwise we have ambiguity. The str type means the path to the file in one case and the contents of the file in other case.

[efolgar]

True, although you can't take advantage of Airflow's built-in macros like task_instance. Also deviates from the behavior of other templated operators (e.g. BashOperator).

[mik-laj]

You can because the templates are rendered recursively.

airflow/airflow/models/baseoperator.py

Lines 872 to 929 in 1c48ffb

	def render_template( # pylint: disable=too-many-return-statements
	self, content: Any, context: Dict, jinja_env: Optional[jinja2.Environment] = None,
	seen_oids: Optional[Set] = None
	) -> Any:
	"""
	Render a templated string. The content can be a collection holding multiple templated strings and will
	be templated recursively.
	:param content: Content to template. Only strings can be templated (may be inside collection).
	:type content: Any
	:param context: Dict with values to apply on templated content
	:type context: dict
	:param jinja_env: Jinja environment. Can be provided to avoid re-creating Jinja environments during
	recursion.
	:type jinja_env: jinja2.Environment
	:param seen_oids: template fields already rendered (to avoid RecursionError on circular dependencies)
	:type seen_oids: set
	:return: Templated content
	"""
	if not jinja_env:
	jinja_env = self.get_template_env()
	# Imported here to avoid ciruclar dependency
	from airflow.models.xcom_arg import XComArg
	if isinstance(content, str):
	if any(content.endswith(ext) for ext in self.template_ext):
	# Content contains a filepath
	return jinja_env.get_template(content).render(**context)
	else:
	return jinja_env.from_string(content).render(**context)
	elif isinstance(content, XComArg):
	return content.resolve(context)
	if isinstance(content, tuple):
	if type(content) is not tuple: # pylint: disable=unidiomatic-typecheck
	# Special case for named tuples
	return content.__class__(
	*(self.render_template(element, context, jinja_env) for element in content)
	)
	else:
	return tuple(self.render_template(element, context, jinja_env) for element in content)
	elif isinstance(content, list):
	return [self.render_template(element, context, jinja_env) for element in content]
	elif isinstance(content, dict):
	return {key: self.render_template(value, context, jinja_env) for key, value in content.items()}
	elif isinstance(content, set):
	return {self.render_template(element, context, jinja_env) for element in content}
	else:
	if seen_oids is None:
	seen_oids = set()
	self._render_nested_template_fields(content, context, jinja_env, seen_oids)
	return content

The problem only occurs when you use Jinja templates to generate new structures e.g. create new array elements.

[mik-laj]

Can you add unit test and some documentation?