threadx

threadx - Create elegant data transformation pipelines. It lets you thread values through a sequence of operations with a sense of clarity and simplicity that feels natural. And it all revolves around two key elements:

• thread: Passes the result of each step as the input to the next.
• x: A smart placeholder that knows exactly where to inject the previous result, whether in a method call, item lookup, or even unpacking.

Here’s what it looks like in action:

from threadx import xf, x

xf('./data.log', 
   read_file, 
   x.splitlines(), 
   (map, x.strip(), x), 
   (map, json.loads, x), 
   (map, x['time'], x), 
   sum)

# or
xl('./data.log', 
   read_file, 
   x.splitlines(), 
   (map, x.strip()), 
   (map, json.loads), 
   (map, x['time']), 
   sum)

What’s happening here? The file content is being read, split, stripped, converted to JSON, and the execution-time summed—all in a linear and readable way. No intermediary variables, no nesting, just the data flowing from one step to the next.

The data.log file (generated by inspector) contains entries like this:

{"time": 12000, "fn": "foo", ...}
{"time": 12345, "fn": "bar", ...}

What Makes threadx Interesting?

• Readable Flow: Instead of diving into layers of nested calls, you write each transformation as a clear, sequential step.
• The x Factor: x acts as a placeholder for where the output of the previous step goes. It’s surprisingly flexible, supporting method calls, attribute/item lookups, and more.
• No Extra Variables: Avoid the noise of intermediate variables or lambda functions. Your transformations stay clean and minimal.

Table of Contents

• Install
• Usage
  • Pass result as first argument
  • Pass x as nth argument
  • Unpacking arguments
  • Method call
  • Attribute lookup
  • Getting Item And Slicing
  • Debugging
  • Fewer lambdas
  • Build data transformation pipeline
• Why I Built This

Install

pip install threadx 

Usage

Import

from threadx import xf, xl, x

Pass result as first argument

xf allows you to pass the result of the previous step automatically as the first argument in each new function:

xf([1, 2, 3],  # => [1, 2, 3]
   sum,        # => 6
   str)        # => '6'

Or, be explicit about it:

xf([1, 2, 3],
   (sum, x),
   (str, x))

Pass x as nth argument

Want to pass the result into a different argument position? No problem:

xf(10, 
   (range, x, 20, 3),  # same as (range, 20, 3)
   list)               # => [10, 13, 16, 19]

xf(20, 
   (range, 10, x, 3),
   list)               # => [10, 13, 16, 19]

xf(3, 
   (range, 10, 20, x),
   list)               # => [10, 13, 16, 19]

Pass x as last argument

xl works same as xf, with just one change, that x will be passed as the last argument.

• xl - pass x as last
• xf - pass x as first

Unpacking arguments

Unpacking works as usual

xf([10, 20], 
   (range, *x, 3),     # unpack to (range, 10, 20, 3)
   list)               # => [10, 13, 16, 19]

Getting Item And Slicing

data = {'a': {'b': [1, 2, 3, 4]}}

xf(data, 
   x['a'], 
   x['b'][0])                   # => 1

xf(data, 
   x['a']['b'][:2])             # => [1, 2]

Attribute lookup

Use x.attribute_name to lookup class and instance attributes.

xf(3 + 4j,
   x.real)                # => 3

xf(3 + 4j,
   (x.real))              # => 3

Method call

Use x.method_name() or x.method_name(args) for method calls, just like magic.

data = {'a': 1, 'b': 2}

xf(data, 
   x.keys(),                 # same as (x.keys())
   list)                     # => ['a', 'b']

xf(data, 
   (x.keys()),
   (list))                    # => ['a', 'b']

xf(data, 
   x.get('c', 'Not Found'))   # => 'Not Found'

Fewer lambdas

Remove verbose lambdas in simple cases.

data = [[1, 2, 3, 4], [10, 20, 30, 40]]

# Normal way:
xf(data, 
   (map, lambda i: i[0], x), 
   list)                                   # => [1, 10]
# or
xf(data, 
   (map, x[0], x), 
   list)                                   # => [1, 10]
# or
xl(data, 
   (map, x[0]), 
   list)                                   # => [1, 10]


# Normal way:
xf(range(12), 
   (filter, lambda i: i % 2 == 0, x), 
   list)                                   # => [0, 2, 4, 6, 8, 10]
# or
xf(range(12), 
   (filter, x % 2 == 0, x), 
   list)                                   # => [0, 2, 4, 6, 8, 10]
# or
xl(range(12), 
   (filter, x % 2 == 0), 
   list)                                   # => [0, 2, 4, 6, 8, 10]

Build data transformation pipeline

# make a tuple or list
pipeline = (read_file, 
            x.splitlines(), 
            (map, x.strip()), 
            (map, json.loads), 
            (map, x['time']), 
            sum)

xl('./data.log', *pipeline)  # works jsut like any other function.

Blowing your brain

# Not saying to solve this problem this way,
# Just showing what `x` can do
answer_sheet =  [{'a': 1, 'b': 2, 'op': op.add    , 'marks': 1, 'answer': 3},
                 {'a': 1, 'b': 2, 'op': op.mul    , 'marks': 2, 'answer': 2},
                 {'a': 1, 'b': 2, 'op': op.truediv, 'marks': 2, 'answer': 0.6} # <- Incorrect answer by student
             ]

# need 60% to pass
passing_marks = 3

# note it is not a lambda
correct_answer = x['op'](x['a'], x['b']) == x['answer']

xl(answer_sheet, 
   (filter, correct_answer), # <--
   (map, x['marks']),
   (sum), 
   (x >= passing_marks)      # <--
  )                          # => True

Why I Built This

After spending a few years working with Clojure, I found myself missing its threading macros when I returned to Python (for a side project). Sure, Python has some tools for chaining operations, but nothing quite as elegant or powerful as what I was used to.