Ansible conditionals in Transilience

This is part of a series of posts on ideas for an ansible-like provisioning system, implemented in Transilience.

I thought a lot of what I managed to do so far with Transilience would be impossible, but then here I am. How about Ansible conditionals? Those must be impossible, right?

Let's give it a try.

A quick recon of Ansible sources

Looking into Ansible's sources, when expressions are lists of strings AND-ed together.

The expressions are Jinja2 expressions that Ansible pastes into a mini-template, renders, and checks the string that comes out.

A quick recon of Jinja2

Jinja2 has a convenient function (jinja2.Environment.compile_expression) that compiles a template snippet into a Python function.

It can also parse a template into an AST that can be inspected in various ways.

Evaluating Ansible conditionals in Python

Environment.compile_expression seems to really do precisely what we need for this, straight out of the box.

There is an issue with the concept of "defined": for Ansible it seems to mean "the variable is present in the template context". In Transilience instead, all variables are fields in the Role dataclass, and can be None when not set.

This means that we need to remove variables that are set to None before passing the parameters to the compiled Jinjae expression:

class Conditional:
    """
    An Ansible conditional expression
    """
    def __init__(self, engine: template.Engine, body: str):
        # Original unparsed expression
        self.body: str = body
        # Expression compiled to a callable
        self.expression: Callable = engine.env.compile_expression(body)

    def evaluate(self, ctx: Dict[str, Any]):
        ctx = {name: val for name, val in ctx.items() if val is not None}
        return self.expression(**ctx)

Generating Python code

Transilience does not only support running Ansible roles, but also converting them to Python code. I can keep this up by traversing the Jinja2 AST generating Python expressions.

The code is straightforward enough that I can throw in a bit of pattern matching to make some expressions more idiomatic for Python:

class Conditional:
    def __init__(self, engine: template.Engine, body: str):
    ...
        parser = jinja2.parser.Parser(engine.env, body, state='variable')
        self.jinja2_ast: nodes.Node = parser.parse_expression()

    def get_python_code(self) -> str:
        return to_python_code(self.jinja2_ast


def to_python_code(node: nodes.Node) -> str:
    if isinstance(node, nodes.Name):
        if node.ctx == "load":
            return f"self.{node.name}"
        else:
            raise NotImplementedError(f"jinja2 Name nodes with ctx={node.ctx!r} are not supported: {node!r}")
    elif isinstance(node, nodes.Test):
        if node.name == "defined":
            return f"{to_python_code(node.node)} is not None"
        elif node.name == "undefined":
            return f"{to_python_code(node.node)} is None"
        else:
            raise NotImplementedError(f"jinja2 Test nodes with name={node.name!r} are not supported: {node!r}")
    elif isinstance(node, nodes.Not):
        if isinstance(node.node, nodes.Test):
            # Special case match well-known structures for more idiomatic Python
            if node.node.name == "defined":
                return f"{to_python_code(node.node.node)} is None"
            elif node.node.name == "undefined":
                return f"{to_python_code(node.node.node)} is not None"
        elif isinstance(node.node, nodes.Name):
            return f"not {to_python_code(node.node)}"
        return f"not ({to_python_code(node.node)})"
    elif isinstance(node, nodes.Or):
        return f"({to_python_code(node.left)} or {to_python_code(node.right)})"
    elif isinstance(node, nodes.And):
        return f"({to_python_code(node.left)} and {to_python_code(node.right)})"
    else:
        raise NotImplementedError(f"jinja2 {node.__class__} nodes are not supported: {node!r}")

Scanning for variables

Lastly, I can implement scanning conditionals for variable references to add as fields to the Role dataclass:

class FindVars(jinja2.visitor.NodeVisitor):
    def __init__(self):
        self.found: Set[str] = set()

    def visit_Name(self, node):
        if node.ctx == "load":
            self.found.add(node.name)


class Conditional:
    ...
    def list_role_vars(self) -> Sequence[str]:
        fv = FindVars()
        fv.visit(self.jinja2_ast)
        return fv.found

The result in action

Take this simple Ansible task:

---
 - name: Example task
   file:
      state: touch
      path: /tmp/test
   when: (is_test is defined and is_test) or debug is defined

Run it through ./provision --ansible-to-python test and you get:

from __future__ import annotations
from typing import Any
from transilience import role
from transilience.actions import builtin, facts


@role.with_facts([facts.Platform])
class Role(role.Role):
    # Role variables used by templates
    debug: Any = None
    is_test: Any = None

    def all_facts_available(self):
        if ((self.is_test is not None and self.is_test)
                or self.debug is not None):
            self.add(
                builtin.file(path='/tmp/test', state='touch'),
                name='Example task')

Besides one harmless set of parentheses too much, what I wasn't sure would be possible is there, right there, staring at me with a mischievous grin.

Next: Building a Transilience playbook in a zipapp.