Skip to content

Commit

Permalink
Create replace_tmpl filter for Jinja2 (#33)
Browse files Browse the repository at this point in the history
  • Loading branch information
DavidNew-NOAA committed Jun 14, 2024
1 parent 8566ea3 commit f6938c7
Show file tree
Hide file tree
Showing 3 changed files with 17 additions and 1 deletion.
3 changes: 3 additions & 0 deletions src/wxflow/jinja.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import os
import sys
from functools import reduce
from pathlib import Path
from typing import Dict, List, Union

Expand Down Expand Up @@ -110,6 +111,7 @@ def get_set_env(self, loader: jinja2.BaseLoader, filters: Dict[str, callable] =
getenv: read variable from environment if defined, else UNDEFINED
to_timedelta: convert a string to a timedelta object
add_to_datetime: add time to a datetime, return new datetime object
replace_tmpl: replace substrings of an input string with replacements specified by an input dictionary
The Expression Statement extension "jinja2.ext.do", which enables
{% do ... %} statements. These are useful for appending to lists.
Expand Down Expand Up @@ -146,6 +148,7 @@ def get_set_env(self, loader: jinja2.BaseLoader, filters: Dict[str, callable] =
if not (isinstance(dt, SilentUndefined) or isinstance(delta, SilentUndefined))
else dt if isinstance(dt, SilentUndefined) else delta)
env.filters["to_timedelta"] = lambda delta_str: to_timedelta(delta_str) if not isinstance(delta_str, SilentUndefined) else delta_str
env.filters["replace_tmpl"] = lambda string, tmpl_dict: reduce(lambda ss, kk: ss.replace(kk, tmpl_dict[kk]), tmpl_dict, string)

# Add any additional filters
if filters is not None:
Expand Down
2 changes: 1 addition & 1 deletion src/wxflow/task.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ def __init__(self, config: Dict, *args, **kwargs):
# Create task_config with everything that is inside _config and whatever the user chooses to
# extend it with when initializing a child subclass of Task. Only task_config should be referenced
# in any application, not _config.
self.task_config = self._config.copy()
self.task_config = self._config.deepcopy()

# Any other composite runtime variables that may be needed for the duration of the task
# can be constructed here
Expand Down
13 changes: 13 additions & 0 deletions tests/test_jinja.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from datetime import datetime

import jinja2
import pytest

from wxflow import Jinja, to_isotime
Expand Down Expand Up @@ -29,6 +30,12 @@ def test_render_stream():
j = Jinja(j2tmpl, data, allow_missing=False)
assert j.render == f"Hello Jane! How are you? It is: {to_isotime(current_date)}"

tmpl_dict = {"{{ name }}": "Jane", "{{ greeting }}": "How are you?", "{{ current_date | to_isotime }}": to_isotime(current_date)}
j = Jinja(j2tmpl, data, allow_missing=False)
loader = jinja2.BaseLoader()
env = j.get_set_env(loader)
assert env.filters['replace_tmpl'](j2tmpl, tmpl_dict) == f"Hello Jane! How are you? It is: {to_isotime(current_date)}"


def test_render_file(tmp_path, create_template):

Expand All @@ -41,6 +48,12 @@ def test_render_file(tmp_path, create_template):
j = Jinja(str(file_path), data, allow_missing=False)
assert j.render == f"Hello Jane! How are you? It is: {to_isotime(current_date)}"

tmpl_dict = {"{{ name }}": "Jane", "{{ greeting }}": "How are you?", "{{ current_date | to_isotime }}": to_isotime(current_date)}
j = Jinja(str(file_path), data, allow_missing=False)
loader = jinja2.BaseLoader()
env = j.get_set_env(loader)
assert env.filters['replace_tmpl'](j2tmpl, tmpl_dict) == f"Hello Jane! How are you? It is: {to_isotime(current_date)}"


def test_include(tmp_path, create_template):

Expand Down

0 comments on commit f6938c7

Please sign in to comment.