Skip to content

Instantly share code, notes, and snippets.

Created April 24, 2019 11:48
Show Gist options
  • Save dliappis/fd1c5ffa2adb74f07bb62f1dc8974aa9 to your computer and use it in GitHub Desktop.
Save dliappis/fd1c5ffa2adb74f07bb62f1dc8974aa9 to your computer and use it in GitHub Desktop.
POC to check for specified track-params that don't alter any of the j2 track params
#!/usr/bin/env python3
import jinja2
import glob
import os
import time
import ipdb
import re
from pathlib import Path
from jinja2 import meta
# See Armin's answer in: for background into why getting the source when includes are used is difficult
base_path = os.path.dirname(template_file_name)
template_vars={'ingest_percentage': 25}
base_track=loader.get_source(jinja2.Environment(), "track.json")
Test this regex interactively in
and use e.g.
{% import "rally.helpers" as rally with context %}\n{\n "version": 2,\n "description": "POIs from Geonames",\n "data-url": "",\n "indices": [\n {\n "name": "geonames",\n "body": "index.json"\n }\n ],\n "corpora": [\n {\n "name": "geonames",\n "base-url": "",\n "documents": [\n {\n "source-file": "documents-2.json.bz2",\n "document-count": 11396505,\n "compressed-bytes": 264698741,\n "uncompressed-bytes": 3547614383\n }\n ]\n }\n ],\n "operations": [\n {{ rally.collect(parts="operations/*.json") }}\n ],\n "challenges": [\n {{ rally.collect(parts="challenges/*.json") }}\n ]\n}\n
as the text string
collect_parts_re = re.compile(r'''{{\ +?rally.collect\(parts="(.+?(?="))"\)\ +?}}''')
source = base_track[0]
match = collect_parts_re.findall(source)
def read_glob_files(pattern):
source = []
files = glob.glob(pattern)
for fname in files:
with open(fname) as fp:
return "".join(source)
if match:
# Construct replacement dict for matched captures
repl = []
for glob_pat in match:
sub_source = read_glob_files(os.path.join(base_path, glob_pat))
# TODO: use recursion instead to replace any {{ rally.collect }} references inside included files too
def replstring(matchobj):
# matchobj.pos has the matched group id
return repl[matchobj.pos]
assembled_source = collect_parts_re.sub(replstring, source)
j2_variables = meta.find_undeclared_variables(ast)
for k in template_vars:
if k not in j2_variables:
print("WARNING: You've declared [{}] track-param but this is not overriding ANY of the j2 variables in the track".format(k))
# now continue rendering the files as we are currently doing in Rally
macros = """
{% macro collect(parts) -%}
{% set comma = joiner() %}
{% for part in glob(parts) %}
{{ comma() }}
{% include part %}
{% endfor %}
{%- endmacro %}
myenv = jinja2.Environment(
loader=jinja2.ChoiceLoader([jinja2.DictLoader({"rally.helpers": macros}), loader])
def relative_glob(start, f):
result = glob.glob(os.path.join(start, f))
if result:
return [os.path.relpath(p, start) for p in result]
return []
glob_helper=lambda f: relative_glob(base_path, f)
# Inject one user supplied track-param for testing
for k, v in template_vars.items():
myenv.globals[k] = v
myenv.globals["now"] = time.time()
myenv.globals["glob"] = glob_helper
rendered_track_json = tmpl.render()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment