**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:
```python
from threadx import thread, x
thread('./data.log',
read_file,
x.splitlines,
(map, x.strip, x),
(map, json.loads, x),
(map, x['time'], x),
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. <br>
The `data.log` file (generated by [inspector](https://github.com/withjak/inspector)) contains entries like this:
```json
{"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](#Install)
- [Usage](#Usage)
- [Pass result as first argument](#Pass-result-as-first-argument)
- [Pass x as nth argument](#Pass-x-as-nth-argument)
- [Unpacking arguments](#Unpacking-arguments)
- [Method call](#Method-call)
- [Attribute lookup](#Attribute-lookup)
- [Getting Item And Slicing](#Getting-Item-And-Slicing)
- [Debugging](#Debugging)
- [Fewer lambdas](#Fewer-lambdas)
- [Build data transformation pipeline](#Build-data-transformation-pipeline)
- [Why I Built This](#Why-I-Built-This)
## Install
```bash
pip install threadx
```
## Usage
### Import
```python
from threadx import thread, x, stop
```
### Pass result as first argument
`thread` allows you to pass the result of the previous step automatically as the first argument in each new function:
```python
thread([1, 2, 3], # => [1, 2, 3]
sum, # => 6
str) # => '6'
```
Or, be explicit about it:
```python
thread([1, 2, 3],
(sum, x),
(str, x))
```
### Pass x as nth argument
Want to pass the result into a different argument position? No problem:
```python
thread(10,
(range, x, 20, 3), # same as (range, 20, 3)
list) # => [10, 13, 16, 19]
thread(20,
(range, 10, x, 3),
list) # => [10, 13, 16, 19]
thread(3,
(range, 10, 20, x),
list) # => [10, 13, 16, 19]
```
### Unpacking arguments
Unpacking works as usual
```python
thread([10, 20],
(range, *x, 3), # unpack to (range, 10, 20, 3)
list) # => [10, 13, 16, 19]
```
### Method call
Use `x.method_name` for method calls, just like magic.
```python
thread(['a', 'b'],
(x.index, 'a')) # => 0
thread(['a', 'b'],
(x.count, 'b')) # => 1
```
### Attribute lookup
Use `x.attribute_name` to lookup class and instance attributes.
```python
thread({'a': 1, 'b': 2},
x.keys,
list) # => ['a', 'b']
```
### Getting Item And Slicing
```python
data = {'a': {'b': [1, 2, 3, 4]}}
thread(data,
x['a'],
x['b'][0]) # => 1
thread(data,
x['a']['b'][:2]) # => [1, 2]
```
### Debugging
Easily inspect intermediate results using `stop`. Usefull for debugging.
```python
thread(data,
x['a'],
x['b'],
stop, # => [1, 2, 3, 4], Stop and return for inspection
sum, # This won’t be executed
str)
```
### Fewer lambdas
Remove verbose lambdas in **simple cases**.
```python
data = [[1, 2, 3, 4], [10, 20, 30, 40]]
# Normal way:
thread(data,
(map, lambda i: i[0], x),
list) # => [1, 10]
# or
thread(data,
(map, x[0], x),
list) # => [1, 10]
# Normal way:
thread(range(12),
(filter, lambda i: i % 2 == 0, x),
list) # => [0, 2, 4, 6, 8, 10]
# or
thread(range(12),
(filter, x % 2 == 0, x),
list) # => [0, 2, 4, 6, 8, 10]
```
### Build data transformation pipeline
```python
# make a tuple or list
pipeline = (read_file,
x.splitlines,
(map, x.strip, x),
(map, json.loads, x),
(map, x['time'], x),
sum)
thread('./data.log', *pipeline) # works jsut like any other function.
```
## 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.
Raw data
{
"_id": null,
"home_page": null,
"name": "threadx",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "chain, chaining, clojure, compose functions, data transformation, function chaining, functional programming, pipe, pipeline, threading",
"author": null,
"author_email": "Akshay Patel <akshaybishnoi@protonmail.com>",
"download_url": "https://files.pythonhosted.org/packages/81/81/90d6c8f24e5d4a64c3866bc05cf821d946358b25e3e3da09809a9d4633bb/threadx-0.1.0a2.tar.gz",
"platform": null,
"description": "**threadx** - Create elegant data transformation pipelines.\nIt 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:\n- **thread**: Passes the result of each step as the input to the next.\n- **x**: A smart placeholder that knows exactly where to inject the previous result, whether in a method call, item lookup, or even unpacking.\n\nHere\u2019s what it looks like in action:\n```python\nfrom threadx import thread, x\n\nthread('./data.log', \n read_file, \n x.splitlines, \n (map, x.strip, x), \n (map, json.loads, x), \n (map, x['time'], x), \n sum)\n```\n\nWhat\u2019s happening here? The file content is being read, split, stripped, converted to JSON, and the execution-time summed\u2014all in a linear and readable way. No intermediary variables, no nesting, just the data flowing from one step to the next. <br>\n\nThe `data.log` file (generated by [inspector](https://github.com/withjak/inspector)) contains entries like this:\n```json\n{\"time\": 12000, \"fn\": \"foo\", ...}\n{\"time\": 12345, \"fn\": \"bar\", ...}\n```\n\nWhat Makes threadx Interesting?\n- **Readable Flow**: Instead of diving into layers of nested calls, you write each transformation as a clear, sequential step. \n- **The `x` Factor**: `x` acts as a placeholder for where the output of the previous step goes. It\u2019s surprisingly flexible, supporting method calls, attribute/item lookups, and more.\n- **No Extra Variables**: Avoid the noise of intermediate variables or lambda functions. Your transformations stay clean and minimal.\n\n# Table of Contents\n- [Install](#Install)\n- [Usage](#Usage)\n - [Pass result as first argument](#Pass-result-as-first-argument)\n - [Pass x as nth argument](#Pass-x-as-nth-argument)\n - [Unpacking arguments](#Unpacking-arguments)\n - [Method call](#Method-call)\n - [Attribute lookup](#Attribute-lookup)\n - [Getting Item And Slicing](#Getting-Item-And-Slicing)\n - [Debugging](#Debugging)\n - [Fewer lambdas](#Fewer-lambdas)\n - [Build data transformation pipeline](#Build-data-transformation-pipeline)\n- [Why I Built This](#Why-I-Built-This)\n\n\n## Install \n```bash\npip install threadx \n```\n\n## Usage\n\n### Import\n```python\nfrom threadx import thread, x, stop\n```\n\n### Pass result as first argument\n`thread` allows you to pass the result of the previous step automatically as the first argument in each new function:\n```python\nthread([1, 2, 3], # => [1, 2, 3]\n sum, # => 6\n str) # => '6'\n```\n\nOr, be explicit about it:\n```python\nthread([1, 2, 3],\n (sum, x),\n (str, x))\n```\n\n### Pass x as nth argument\nWant to pass the result into a different argument position? No problem:\n```python\nthread(10, \n (range, x, 20, 3), # same as (range, 20, 3)\n list) # => [10, 13, 16, 19]\n\nthread(20, \n (range, 10, x, 3),\n list) # => [10, 13, 16, 19]\n\nthread(3, \n (range, 10, 20, x),\n list) # => [10, 13, 16, 19]\n```\n\n### Unpacking arguments \n\nUnpacking works as usual\n\n```python \nthread([10, 20], \n (range, *x, 3), # unpack to (range, 10, 20, 3)\n list) # => [10, 13, 16, 19]\n```\n\n### Method call\nUse `x.method_name` for method calls, just like magic.\n```python\nthread(['a', 'b'], \n (x.index, 'a')) # => 0\n\nthread(['a', 'b'], \n (x.count, 'b')) # => 1\n```\n\n### Attribute lookup\nUse `x.attribute_name` to lookup class and instance attributes. \n```python \nthread({'a': 1, 'b': 2},\n x.keys, \n list) # => ['a', 'b']\n\n```\n\n### Getting Item And Slicing\n```python\ndata = {'a': {'b': [1, 2, 3, 4]}}\n\nthread(data, \n x['a'], \n x['b'][0]) # => 1\n\nthread(data, \n x['a']['b'][:2]) # => [1, 2]\n\n```\n\n### Debugging \nEasily inspect intermediate results using `stop`. Usefull for debugging.\n```python\nthread(data, \n x['a'], \n x['b'], \n stop, # => [1, 2, 3, 4], Stop and return for inspection\n sum, # This won\u2019t be executed\n str)\n\n```\n\n### Fewer lambdas\nRemove verbose lambdas in **simple cases**.\n```python \ndata = [[1, 2, 3, 4], [10, 20, 30, 40]]\n\n# Normal way:\nthread(data, \n (map, lambda i: i[0], x), \n list) # => [1, 10]\n# or\nthread(data, \n (map, x[0], x), \n list) # => [1, 10]\n\n\n# Normal way:\nthread(range(12), \n (filter, lambda i: i % 2 == 0, x), \n list) # => [0, 2, 4, 6, 8, 10]\n# or\nthread(range(12), \n (filter, x % 2 == 0, x), \n list) # => [0, 2, 4, 6, 8, 10]\n```\n\n### Build data transformation pipeline\n```python\n# make a tuple or list\npipeline = (read_file, \n x.splitlines, \n (map, x.strip, x), \n (map, json.loads, x), \n (map, x['time'], x), \n sum)\n\nthread('./data.log', *pipeline) # works jsut like any other function.\n```\n\n## Why I Built This\nAfter 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.\n\n\n\n\n\n\n",
"bugtrack_url": null,
"license": null,
"summary": "Chaining function calls on steroids",
"version": "0.1.0a2",
"project_urls": {
"Homepage": "https://github.com/withjak/threadx",
"Issues": "https://github.com/withjak/threadx/issues"
},
"split_keywords": [
"chain",
" chaining",
" clojure",
" compose functions",
" data transformation",
" function chaining",
" functional programming",
" pipe",
" pipeline",
" threading"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "a0c71eb18906570d0257e3d618e828b1711a273b20461d5304f219cbada79116",
"md5": "33ed041a65be603857809ef8e03852b3",
"sha256": "248567afc95d7d01aa3557ee7235bbd26df6d91fbaf45e2fc85193dbb7e92a28"
},
"downloads": -1,
"filename": "threadx-0.1.0a2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "33ed041a65be603857809ef8e03852b3",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 9860,
"upload_time": "2024-09-22T04:10:10",
"upload_time_iso_8601": "2024-09-22T04:10:10.200179Z",
"url": "https://files.pythonhosted.org/packages/a0/c7/1eb18906570d0257e3d618e828b1711a273b20461d5304f219cbada79116/threadx-0.1.0a2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "818190d6c8f24e5d4a64c3866bc05cf821d946358b25e3e3da09809a9d4633bb",
"md5": "148203a4cabe3be1c56e08a32b70f524",
"sha256": "5728d4716092c02efec166466b3c6d2940ceb1086ef8c986b0fe1828a7956038"
},
"downloads": -1,
"filename": "threadx-0.1.0a2.tar.gz",
"has_sig": false,
"md5_digest": "148203a4cabe3be1c56e08a32b70f524",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 10739,
"upload_time": "2024-09-22T04:10:11",
"upload_time_iso_8601": "2024-09-22T04:10:11.913933Z",
"url": "https://files.pythonhosted.org/packages/81/81/90d6c8f24e5d4a64c3866bc05cf821d946358b25e3e3da09809a9d4633bb/threadx-0.1.0a2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-09-22 04:10:11",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "withjak",
"github_project": "threadx",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "threadx"
}
JSON
