Dave Fancher: The Book of F#



book cover Few months ago I decided to try something different, like different platform and different language. As I mostly work with Python/JS/JVM langs, I chose .NET and F#. So I read The Book of F# by Dave Fancher from Joy of Coding bundle. And it was interesting. In the book aspects of F# are nicely explained with examples. Also book covers non-straightforward parts of the language, and explains why it was made that way.

Although author of the book assumed that readers would be familiar with .NET platform and C#, it wasn’t a problem at all.

Site Reliability Engineering: How Google Runs Production Systems



book cover white Not so long ago I wanted to read something about DevOps and processes in real organisations. So I chose Site Reliability Engineering: How Google Runs Production Systems. And it nicely explains about deploy, failures recovery, support and other SRE aspects from engineering and management points of view. Also it interesting to read about problems and solutions of huge systems.

However the book is informative, but it’s a bit boring. And almost most of cases are Google specific or can be a problem only on a very large systems.

AST transformations with __future__-like module



In the previous article I wrote how-to add partial application with ... and piping with @ using AST transformations. However we needed to transform AST manually. For automatizing it I planned to use macropy but it doesn’t work with Python 3 and a bit too complicated. So I ended up with an idea to create __transformers__ module that work in a similar way with Python’s __future__ module. So code will look like:

from __transformers__ import ellipsis_partial, matmul_pipe


range(10) @ map(lambda x: x ** 2, ...) @ list @ print

So first of all for implementing it we need to extract enabled transformers names from code, it’s easy with ast.NodeVisitor, we just process all ImportForm nodes:

import ast


class NodeVisitor(ast.NodeVisitor):
    def __init__(self):
        self._found = []

    def visit_ImportFrom(self, node):
        if node.module == '__transformers__':
            self._found += [name.name for name in node.names]

    @classmethod
    def get_transformers(cls, tree):
        visitor = cls()
        visitor.visit(tree)
        return visitor._found

Let’s run it:

tree = ast.parse(code)

>>> print(NodeVisitor.get_transformers(tree))
['ellipsis_partial', 'matmul_pipe']

Next step is to define transformers. Transformer is just a Python module with transformer variable, that is instance of ast.NodeTransformer. For example transformer module for piping with matrix multiplication operator will be like:

import ast


class MatMulPipeTransformer(ast.NodeTransformer):
    def _replace_with_call(self, node):
        """Call right part of operation with left part as an argument."""
        return ast.Call(func=node.right, args=[node.left], keywords=[])

    def visit_BinOp(self, node):
        if isinstance(node.op, ast.MatMult):
            node = self._replace_with_call(node)
            node = ast.fix_missing_locations(node)

        return self.generic_visit(node)


transformer = MatMulPipeTransformer()

Now we can write function that extracts used transformers, imports and applies it to AST:

def transform(tree):
    transformers = NodeVisitor.get_transformers(tree)

    for module_name in transformers:
        module = import_module('__transformers__.{}'.format(module_name))
        tree = module.transformer.visit(tree)

    return tree

And use it on our code:

from astunparse import unparse

>>> unparse(transform(tree))
from __transformers__ import ellipsis_partial, matmul_pipe
print(list((lambda __ellipsis_partial_arg_0: map((lambda x: (x ** 2)), __ellipsis_partial_arg_0))(range(10)))

Next part is to automatically apply transformations on module import, for that we need to implement custom Finder and Loader. Finder is almost similar with PathFinder, we just need to replace Loader with ours in spec. And Loader is almost SourceFileLoader, but we need to run our transformations in source_to_code method:

from importlib.machinery import PathFinder, SourceFileLoader


class Finder(PathFinder):
    @classmethod
    def find_spec(cls, fullname, path=None, target=None):
        spec = super(Finder, cls).find_spec(fullname, path, target)
        if spec is None:
            return None

        spec.loader = Loader(spec.loader.name, spec.loader.path)
        return spec


class Loader(SourceFileLoader):
    def source_to_code(self, data, path, *, _optimize=-1):
        tree = ast.parse(data)
        tree = transform(tree)
        return compile(tree, path, 'exec',
                       dont_inherit=True, optimize=_optimize)

Then we need to put our finder in sys.meta_path:

import sys

def setup():
    sys.meta_path.insert(0, Finder)
    
setup()

And now we can just import modules that use transformers. But it requires some bootstrapping.

We can make it easier by creating __main__ module that will register module finder and run module or file:

from runpy import run_module
from pathlib import Path
import sys
from . import setup

setup()

del sys.argv[0]

if sys.argv[0] == '-m':
    del sys.argv[0]
    run_module(sys.argv[0])
else:
    # rnupy.run_path ignores meta_path for first import
    path = Path(sys.argv[0]).parent.as_posix()
    module_name = Path(sys.argv[0]).name[:-3]
    sys.path.insert(0, path)
    run_module(module_name)

So now we can run our module easily:

➜ python -m __transformers__ -m test   
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

➜ python -m __transformers__ test.py                 
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

And that’s all, you can try transformers by yourself with transformers package:

pip install transformers

Source code on github, package, previous part.

Partial application and piping with AST transformation



In the previous article I wrote about how to implement partial application and piping using operator overloading and decorators. But we can use a bit different approach – AST transformation.

For example we have code:

def add(x, y):
    return x + y
    
    
addFive = add(..., 5)

print(addFive(10))

We can look to AST of this code using ast module from standard library and dump function from gist:

import ast

code = open('src.py')  # the previous code
tree = ast.parse(code)
print(dump(tree))

It would be like:

Module(body=[
    FunctionDef(name='add', args=arguments(args=[
        arg(arg='x', annotation=None),
        arg(arg='y', annotation=None),
      ], vararg=None, kwonlyargs=[], kw_defaults=[], kwarg=None, defaults=[]), body=[
        Return(value=BinOp(left=Name(id='x', ctx=Load()), op=Add(), right=Name(id='y', ctx=Load()))),
      ], decorator_list=[], returns=None),
    Assign(targets=[
        Name(id='addFive', ctx=Store()),
      ], value=Call(func=Name(id='add', ctx=Load()), args=[
        Ellipsis(),
        Num(n=5),
      ], keywords=[])),
    Expr(value=Call(func=Name(id='print', ctx=Load()), args=[
        Call(func=Name(id='addFive', ctx=Load()), args=[
            Num(n=10),
          ], keywords=[]),
      ], keywords=[])),
  ])

And we can easily spot call with ellipsis argument:

Call(func=Name(id='add', ctx=Load()), args=[
    Ellipsis(),
    Num(n=5),
  ], keywords=[])

We need to wrap each call with ellipsis in lambda and replace ... with the lambda’s argument. We can do it with ast.NodeTransformer. It calls visit_Call method for each Call node:

class EllipsisPartialTransform(ast.NodeTransformer):
    def __init__(self):
        self._counter = 0

    def _get_arg_name(self):
        """Return unique argument name for lambda."""
        try:
            return '__ellipsis_partial_arg_{}'.format(self._counter)
        finally:
            self._counter += 1

    def _is_ellipsis(self, arg):
        return isinstance(arg, ast.Ellipsis)

    def _replace_argument(self, node, arg_name):
        """Replace ellipsis with argument."""
        replacement = ast.Name(id=arg_name,
                               ctx=ast.Load())
        node.args = [replacement if self._is_ellipsis(arg) else arg
                     for arg in node.args]
        return node

    def _wrap_in_lambda(self, node):
        """Wrap call in lambda and replace ellipsis with argument."""
        arg_name = self._get_arg_name()
        node = self._replace_argument(node, arg_name)
        return ast.Lambda(
            args=ast.arguments(args=[ast.arg(arg=arg_name, annotation=None)],
                               vararg=None, kwonlyargs=[], kw_defaults=[],
                               kwarg=None, defaults=[]),
            body=node)

    def visit_Call(self, node):
        if any(self._is_ellipsis(arg) for arg in node.args):
            node = self._wrap_in_lambda(node)
            node = ast.fix_missing_locations(node)

        return self.generic_visit(node)

So now we can transform AST with visit method and dump result:

tree = EllipsisPartialTransform().visit(tree)
print(dump(tree))

And you can see changes:

Module(body=[
    FunctionDef(name='add', args=arguments(args=[
        arg(arg='x', annotation=None),
        arg(arg='y', annotation=None),
      ], vararg=None, kwonlyargs=[], kw_defaults=[], kwarg=None, defaults=[]), body=[
        Return(value=BinOp(left=Name(id='x', ctx=Load()), op=Add(), right=Name(id='y', ctx=Load()))),
      ], decorator_list=[], returns=None),
    Assign(targets=[
        Name(id='addFive', ctx=Store()),
      ], value=Lambda(args=arguments(args=[
        arg(arg='__ellipsis_partial_arg_0', annotation=None),
      ], vararg=None, kwonlyargs=[], kw_defaults=[], kwarg=None, defaults=[]), body=Call(func=Name(id='add', ctx=Load()), args=[
        Num(n=5),
        Name(id='__ellipsis_partial_arg_0', ctx=Load()),
      ], keywords=[]))),
    Expr(value=Call(func=Name(id='print', ctx=Load()), args=[
        Call(func=Name(id='addFive', ctx=Load()), args=[
            Num(n=10),
          ], keywords=[]),
      ], keywords=[])),
  ])

AST is not easy to read, so we can use astunparse for transforming it to source code:

from astunparse import unparse

print(unparse(tree))

Result is a bit ugly, but more readable than AST:

def add(x, y):
    return (x + y)
addFive = (lambda __ellipsis_partial_arg_0: add(5, __ellipsis_partial_arg_0))
print(addFive(10))

For testing result we can compile AST and run it:

exec(compile(tree, '<string>', 'exec'))
#  15

And it’s working! Back to piping, for example we have code:

"hello world" @ str.upper @ print

It’s AST would be:

Module(body=[
    Expr(value=BinOp(left=BinOp(left=Str(s='hello world'),
                     op=MatMult(),
                     right=Attribute(value=Name(id='str', ctx=Load()), attr='upper', ctx=Load())), 
                                     op=MatMult(),
                                     right=Name(id='print', ctx=Load()))),
  ])

BinOp with op=MatMult() is place where we use matrix multiplication operator. We need to transform it to call of right part with left part as an argument:

class MatMulPipeTransformation(ast.NodeTransformer):
    def _replace_with_call(self, node):
        """Call right part of operation with left part as an argument."""
        return ast.Call(func=node.right, args=[node.left], keywords=[])

    def visit_BinOp(self, node):
        if isinstance(node.op, ast.MatMult):
            node = self._replace_with_call(node)
            node = ast.fix_missing_locations(node)

        return self.generic_visit(node)

Transformed AST would be:

Module(body=[
    Expr(value=Call(func=Name(id='print', ctx=Load()), args=[
        Call(func=Attribute(value=Name(id='str', ctx=Load()), attr='upper', ctx=Load()), args=[
            Str(s='hello world'),
          ], keywords=[]),
      ], keywords=[])),
  ])

And result code is just a nested calls:

print(str.upper('hello world'))
#  HELLO WORLD

So now it’s time to combine both transformers. For example we have code:

from functools import reduce
import operator

range(100) @ filter(lambda x: x % 2 == 0, ...) \
           @ map(lambda x: x ** 2, ...) \
           @ zip(..., range(200, 250)) \
           @ map(sum, ...) \
           @ reduce(operator.add, ...) \
           @ str.format('result: {}', ...) \
           @ str.upper \
           @ print

We can transform and run it with:

code = open('src.py')  # the previous code
tree = ast.parse(code)

tree = MatMulPipeTransformation().visit(
    EllipsisPartialTransform().visit(tree))
    
exec(compile(tree, '<string>', 'exec'))

It’s working, output as expected is:

RESULT: 172925

However result code is a bit messy:

from functools import reduce
import operator
print(str.upper((lambda __ellipsis_partial_arg_5: str.format('result: {}', __ellipsis_partial_arg_5))(
    (lambda __ellipsis_partial_arg_4: reduce(operator.add, __ellipsis_partial_arg_4))(
        (lambda __ellipsis_partial_arg_3: map(sum, __ellipsis_partial_arg_3))(
            (lambda __ellipsis_partial_arg_2: zip(__ellipsis_partial_arg_2, range(200, 250)))(
                (lambda __ellipsis_partial_arg_1: map((lambda x: (x ** 2)), __ellipsis_partial_arg_1))(
                    (lambda __ellipsis_partial_arg_0: filter((lambda x: ((x % 2) == 0)), __ellipsis_partial_arg_0))(
                        range(100)))))))))

This approach is better then previous, we don’t need to manually wrap all functions with ellipsis_partial or use _ helper. Also we don’t use custom Partial. But with this approach we need to manually transform AST, so in the next part I’ll show how we can do it automatically with module finder/loader.

Gist with sources, previous part, next part.

Partial application and piping with ... and @



In newer versions of Python we have two not much used features: ellipsis:

>>> print(...)
Ellipsis

And matrix multiplication operator:

class Dummy(str):
    def __matmul__(self, other):
        print('{}@{}'.format(self, other))


>>> Dummy('ok') @ 'there'
ok@there

So let’s start with ..., in Scala we can partially apply (or curry) function with _:

def add(x: Int, y: Int) = x + y
val addOne = add(1, _: Int)
addOne(5)  6: Int

Wouldn’t it be nice to have similar option in Python, like:

def add(x, y):
    return x + y
    
addFive = add(..., 5)

And we can easily implement it with some decorator:

from functool import wraps

class Partial:
    def __init__(self, fn, args, kwargs):
        self._fn = fn
        self._args = args
        self._kwargs = kwargs

    def __call__(self, replacement):
        args = [replacement if arg is ... else arg
                for arg in self._args]
        kwargs = {key: replacement if val is ... else val
                  for key, val in self._kwargs.items()}
        return self._fn(*args, **kwargs)

    def __repr__(self):
        return '<Partial: {}(*{}, **{})>'.format(
            self._fn.__name__, repr(self._args), repr(self._kwargs))


def ellipsis_partial(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        ellipsises = (list(args) + list(kwargs.values())).count(...)
        if ellipsises > 1:
            raise TypeError('Only one ... allowed as an argument.')
        elif ellipsises:
            return Partial(fn, args, kwargs)
        else:
            return fn(*args, **kwargs)

    return wrapper

So here if we find ... in arguments, we return Partial object. And when the object called – we replace ... with passed value. In action:

@ellipsis_partial
def add(x, y):
    return x + y


addFive = add(5, ...)
>>> addFive(10)
15

And it works! So back to matrix multiplication operator. In F# there’s nice piping operators:

> [1..10] |> List.filter (fun x -> x % 3 = 0)
val it : int list = [3; 6; 9]

So with our operator in Python it should look like:

range(1, 10) @ filter(lambda x: x % 3 == 0, ...)

And we can easily implement it just by adding __rmatmul__ to Partial:

class Partial:
    def __init__(self, fn, args, kwargs):
        self._fn = fn
        self._args = args
        self._kwargs = kwargs

    def __call__(self, replacement):
        args = [replacement if arg is ... else arg
                for arg in self._args]
        kwargs = {key: replacement if val is ... else val
                  for key, val in self._kwargs.items()}
        return self._fn(*args, **kwargs)

    def __rmatmul__(self, replacement):
        return self(replacement)

    def __repr__(self):
        return '<Partial: {}(*{}, **{})>'.format(
            self._fn.__name__, repr(self._args), repr(self._kwargs))

And in action:

filter = ellipsis_partial(filter)
to_list = ellipsis_partial(list)
>>> range(1, 10) @ filter(lambda x: x % 3 == 0, ...) @ to_list(...)
[3, 6, 9]

And we can use it in even more complex cases:

map = ellipsis_partial(map)
join = ellipsis_partial(str.join)
>>> range(1, 10) @ map(lambda x: x + 4, ...) \
                 @ filter(lambda x: x % 3 == 0, ...) \
                 @ map(str, ...) \
                 @ join(', ', ...)
6, 9, 12

But it’s a bit not nice to wrap all callables in ellipsis_partial, we can use some hacks with inspect or module loaders to doing it automatically, but it’s too magic for me. So we can add little function that wrap and call:

def _(fn, *args, **kwargs):
    return ellipsis_partial(fn)(*args, **kwargs)

Usage:

from functools import reduce
>>> range(1, 10) @ map(lambda x: x + 4, ...) \
                 @ filter(lambda x: x % 3 == 0, ...) \
                 @ _(reduce, lambda x, y: x * y, ...) \
                 @ _('value: {}'.format, ...)
value: 648

However it may look strange and unpythonic, but I guess it would be nice to see something like this in future Python releases.

Gist with sources, next part.

Abusing annotations with dependency injection



Python 3 has a nice feature – type annotations:

def add(x: int, y: int) -> int:
    return x + y

That can be used by IDEs and stuff like mypy for type checking. However we can easily access it:

>>> add.__annotations__
{'return': int, 'x': int, 'y': int}

And use it for things like dependency injection. For example we have a web app:

def get_db_connection() -> abc.DBConnection:
    ...


def get_routes() -> abc.Router:
    ...


def get_cache(db: abc.DBConnection) -> abc.CacheManager:
    ...


def init_app():
    db = get_db_connection()
    routes = get_routes()
    cache = get_cache(db)
    app = Application(routes=routes,
                      db=db,
                      cache=cache)
    app.run()


if __name__ == '__main__':
    init_app()

Looks a bit Java-like with interfaces (abstract classes, abc), but it’s useful in huge apps. However components are tightly coupled, and we need to use monkey patching for testing it.

Let’s examine annotations:

>>> get_cache.__annotations__
{'db': abc.DBConnection, 'return': abc.CacheManager}

We can see that the function requires abc.DBConnection and provides abc.CacheManager. We need to track all functions like this, it’ll be easy with some decorator:

from weakref import WeakValueDictionary

_provides = WeakValueDictionary()


def provides(fn):
    """Register function that provides something."""
    try:
        _provides[fn.__annotations__['return']] = fn
    except KeyError:
        raise ValueError('Function not annotated.')

    return fn

We use WeakValueDictionary in case function somehow can be deleted.

Let’s apply this decorator:

@provides
def get_db_connection() -> abc.DBConnection:
    ...


@provides
def get_routes() -> abc.Router:
    ...


@provides
def get_cache(*, db: abc.DBConnection) -> abc.CacheManager:
    ...

And move dependencies of main function to arguments:

def init_app(*, routes: abc.Router,
             db: abc.DBConnection,
             cache: abc.CacheManager):
    app = Application(routes=routes,
                      db=db,
                      cache=cache)
    app.run()

So we can think about our functions as a graph:

graph TB A[init_app]---B[get_routes] A---C[get_db_connection] A---D[get_cache] D---C

And we can easily write injector that resolve and inject dependencies:

class Injector:
    """Resolve and inject dependencies."""

    def __init__(self):
        self._resolved = {}

    def _get_value(self, name):
        """Get dependency by name (type)."""
        if name not in _provides:
            raise ValueError("Dependency {} not registered.".format(
                name))
        
        if name not in self._resolved:           
            fn = _provides[name]
            kwargs = self._get_dependencies(fn)
            return fn(**kwargs)
        return self._resolved[name]

    def _get_dependencies(self, fn):
        """Get dependencies for function."""
        return {key: self._get_value(value)
                for key, value in fn.__annotations__.items()
                if key != 'return'}

    def run(self, fn):
        """Resolve dependencies and run function."""
        kwargs = self._get_dependencies(fn)
        return fn(**kwargs)

So we can make our app work by adding:

if __name__ == '__main__':
    Injector().run(init_app)

Although this approach is simple and straightforward, it’s overkill for most of apps.

Package on github.

Rerenderer rendering performance



Rerenderer is a React-like library for cross platform drawing on canvas. And we experimenting a lot with ways to improve performance of rendering. Not so while ago I wrote about intermediate language and interpreter, this approach was interesting, but not so much efficient. Performing optimizations on that language wasn’t enough fast and interpreter on Android was based on reflection and was a bit slow because of that.

So now we decided to sacrifice flexibility and move component implementation to host platforms (we use Kotlin for Android and ClojureScript for browsers). And use simpler approach with canvases tree (one canvas for each component), where we rerender only changed canvases and ancestors. So for example we have a simple app:

(ns rerenderer.example.core
  (:require [rerenderer.core :refer [init!]]
            [rerenderer.primitives :refer [rectangle text]]
            [rerenderer.debug :refer [swap-state!]]))

(defn labeled
  [{:keys [label width] :as options} & children]
  (rectangle options
    children
    (text {:x (- width 50) :y 0
           :width 40 :height 40
           :font-size 36
           :color "#3E454C"
           :value label})))

(defn root
  [{:keys [background-color first-color second-color third-color]}]
  (labeled {:x 0 :y 0
            :width 800 :height 400
            :color background-color
            :label "A"}
    (labeled {:x 30 :y 30
              :width 200 :height 200
              :color first-color
              :label "B"})
    (labeled {:x 100 :y 100
              :width 500 :height 250
              :color second-color
              :label "C"}
      (labeled {:x 400 :y 150
                :width 100 :height 100
                :color third-color
                :label "D"}))))

(def initial-state
  {:background-color "#FFF6E5"
   :first-color "#7ECEFD"
   :second-color "#FF7F66"
   :third-color "#2185C5"})

(defonce app (init! :root-view #'root
                    :state initial-state
                    :canvas (.getElementById js/document "canvas")
                    :width 800
                    :height 600))

Whole scene can be represented as a tree:

graph TB A[A]-->B[B] A-->C[C] C-->D[D]

So when we change D:

(swap-state! app assoc :third-color :red)

We rerender D, C and A, but don’t touch B:

graph TB A[A]-->B[B] A-->C[C] C-->D[D] style D fill:red; style C stroke:red; style A stroke:red;

But when we change C:

(swap-state! app assoc :second-color :white)

We rerender only C, A and don’t touch B and D. Because changes doesn’t affect it canvases:

graph TB A[A]-->B[B] A-->C[C] C-->D[D] style C fill:red; style A stroke:red;

You can easily notice performance boost in browser on more complex example – Conway’s Game of Life:

And on Android:

It’s faster then before, but you can notice that it’s not smooth, now we have a little problem with GC.

Configure Jest for React Native



Jest is a very popular unit testing tool for React, but it doesn’t work with React Native out of the box. And even instruction in React Native docs is wrong. If you use it you’ll end up without ability to disable mocks, without working es6 imports and with ton of ReferenceError.

So, I accumulated information from Jest issues and got working config.

First of all we need to install jest-cli and babel-jest:

npm install --save-dev jest-cli babel-jest babel-polyfill babel-preset-react-native

Then fill .babelrc with:

{
  "presets": ["react-native"],
  "retainLines": true
}

Without retainLines you’ll get wrong line numbers in tests traces.

And update package.json:

{
  ...
  "scripts": {
    ...
    "test": "jest"
  },
  "jest": {
    "haste": {
      "defaultPlatform": "android",
      "platforms": [
        "ios",
        "android"
      ],
      "providesModuleNodeModules": [
        "react-native"
      ]
    },
    "unmockedModulePathPatterns": [
      "promise",
      "source-map"
    ]
  }
}

That’s all, now you can run tests with:

npm test

Also a lot of outdated docs uses jest.autoMockOff, it’s deprecated and doesn’t work with es6 imports. You should use jest.disableAutomock().

If you want to use use enzyme for testing React Native components, you should install:

npm install --save-dev enzyme react-addons-test-utils react-dom react-native-mock

Then add react and enzyme to unmockedModulePathPatterns. And create __mocks__/react-native.js with:

import ReactNative from 'react-native-mock';

__DEV__ = true;

module.exports = ReactNative;

Now you can mock React Native and use enzyme.

Parse shell one-liners with pyparsing



For one of my projects I needed some one-liners parser to AST. I’ve tried PLY, pyPEG and a few more. And stopped on pyparsing. It’s actively maintained, works without magic and easy to use.

Ideally I wanted to parse something like:

LANG=en_US.utf-8 git diff | wc -l >> diffs

To something like:

(= LANG en_US.utf-8)
(>> (| (git diff) (wc -l))
    (diffs))

So let’s start with simple shell command, it’s just space-separated tokens:

import pyparsing as pp


token = pp.Word(pp.alphanums + '_-.')
command = pp.OneOrMore(token)

command.parseString('git branch --help')
>>> ['git', 'branch', '--help']

It’s simple, another simple part is parsing environment variables. One environment variable is token=token, and list of them separated by spaces:

env = pp.Group(token + '=' + token)

env.parseString('A=B')
>>>[['A', '=', 'B']]

env_list = pp.OneOrMore(env)

env_list.parseString('VAR=test X=1')
>>> [['VAR', '=', 'test'], ['X', '=', '1']]

And now we can easily merge command and environment variables, mind that environment variables are optional:

command_with_env = pp.Optional(pp.Group(env_list)) + pp.Group(command)

command_with_env.parseString('LOCALE=en_US.utf-8 git diff')
>>> [[['LOCALE', '=', 'en_US.utf-8']], ['git', 'diff']]

Now we need to add support of pipes, redirects and logical operators. Here we don’t need to know what they’re doing, so we’ll treat them just like separators between commands:

separators = ['1>>', '2>>', '>>', '1>', '2>', '>', '<', '||', '|', '&&', '&', ';']
separator = pp.oneOf(separators)
command_with_separator = pp.OneOrMore(pp.Group(command) + pp.Optional(separator))

command_with_separator.parseString('git diff | wc -l >> out.txt')
>>> [['git', 'diff'], '|', ['wc', '-l'], '>>', ['out.txt']]

And now we can merge environment variables, commands and separators:

one_liner = pp.Optional(pp.Group(env_list)) + pp.Group(command_with_separator)
            
one_liner.parseString('LANG=C DEBUG=true git branch | wc -l >> out.txt')
>>> [[['LANG', '=', 'C'], ['DEBUG', '=', 'true']], [['git', 'branch'], '|', ['wc', '-l'], '>>', ['out.txt']]]

Result is hard to process, so we need to structure it:

one_liner = pp.Optional(env_list).setResultsName('env') + \
            pp.Group(command_with_separator).setResultsName('command')
result = one_liner.parseString('LANG=C DEBUG=true git branch | wc -l >> out.txt')

print('env:', result.env, '\ncommand:', result.command)
>>> env: [['LANG', '=', 'C'], ['DEBUG', '=', 'true']] 
>>> command: [['git', 'branch'], '|', ['wc', '-l'], '>>', ['out.txt']]

Although we didn’t get AST, but just a bunch of grouped tokens. So now we need to transform it to proper AST:

def prepare_command(command):
    """We don't need to work with pyparsing internal data structures,
    so we just convert them to list.
    
    """
    for part in command:
        if isinstance(part, str):
            yield part
        else:
            yield list(part)


def separator_position(command):
    """Find last separator position."""
    for n, part in enumerate(command[::-1]):
        if part in separators:
            return len(command) - n - 1


def command_to_ast(command):
    """Recursively transform command to AST.""" 
    n = separator_position(command)
    if n is None:
        return tuple(command[0])
    else:
        return (command[n],
                command_to_ast(command[:n]),
                command_to_ast(command[n + 1:]))


def to_ast(parsed):
    if parsed.env:
        for env in parsed.env:
            yield ('=', env[0], env[2])
    command = list(prepare_command(parsed.command))
    yield command_to_ast(command)
   
   
list(to_ast(result))
>>> [('=', 'LANG', 'C'),
>>>  ('=', 'DEBUG', 'true'),
>>>  ('>>', ('|', ('git', 'branch'),
>>>               ('wc', '-l')),
>>>         ('out.txt',))]

It’s working. The last part, glue that make it easier to use:

def parse(command):
    result = one_liner.parseString(command)
    ast = to_ast(result)
    return list(ast)
    
    
parse('LANG=en_US.utf-8 git diff | wc -l >> diffs')
>>> [('=', 'LANG', 'en_US.utf-8'),
     ('>>', ('|', ('git', 'diff'),
                  ('wc', '-l')),
            ('diffs',))]

Although it can’t parse all one-liners, it doesn’t support nested commands like:

echo $(git branch)
echo `git branch`

But it’s enough for my task and support of not implemented features can be added easily.

Gist with source code.

Three months in Cyprus



Nicosia aqueduct

From March to May I was in Cyprus. Most of time I was in Nicosia, but visited most of Cyprus’s cities. Nicosia located in the middle of the island, 30-50 minutes by bus from the sea (Larnaca). The city split between two countries and has some sort of wall.

For entering country I was using Pro Visa, it’s very easy to obtain, you just need to fill some form and send to consulate by email. It allows to stay 90 days in Cyprus.

In Nicosia I was living in city center, and was renting apartment with shared kitchen for €300 including water, electricity and internet. Internet was like 20 MBit/s.

Mobile internet is a bit pricey, I was using MTN and paid €15 for 1.5GB for month. LTE available in almost all cities.

Food is relatively cheap, for everyday meat/fish/seafood I spend around €40 per week. Prices is supermarket near my apartment was €4-5 for pork, €5-6 for chicken, €5-10 for fish and €8-10 for octopus and other seafood, €0.3-1 for vegetables.

Alcohol is also cheap, €1-1.5 for can of beer, €4-5 for six pack. Local wine was around €4-5. Local strong drinks like Ouzo and Zivania was €5-10.

Main problem with shopping in Cyprus, is that shops close very early on Wednesday and Saturday, and they doesn’t work on Sundays.

Although, Nicosia is very comfortable place to live.

In this trip I also visited Israel for two weeks, lived in Tel-Aviv. And Georgia, for two weeks too, lived in Tbilisi. They’re both very nice and interesting places.