python-pods

A faithful port of the babashka pods library to python.

python-pods allow interop with all pods that implement the pod protocol defined by babashka pods. You can load and run any pod from the pod registry.

python pods has a 'patch system' to override certain behaviour exposed by pods which expect the client to be clojure/ babashka. Details below.

Features

Load and communicate with pods using EDN, JSON, or Transit+JSON formats
Automatic pod downloading from the babashka pod registry
Patch existing pods to make more python friendly when necessary
Expose pod namespaces as importable Python modules
Support for custom EDN readers and Transit transforms
Metadata preservation with Transit+JSON format
Dynamic registration of custom Transit read/write handlers
Automatic type conversion between Python and pod data types
Thread-safe communication with pods

Installation

This project uses uv as the Python package manager for fast and reliable dependency management.

pip install python-pods
# or
uv add python-pods

Quick Start

The instaparse pod is clojure pod in the babashka pod registry for working with the wonderful instaparse library. Let's use it in python!

import python_pods as pods

# Load a pod from the pod registry
pod = pods.load_pod('org.babashka/instaparse', {'version': '0.0.6'})

# Import the pod namespace as a Python module
import pod_babashka_instaparse as insta

# Use functions from the pod
parser = insta.parser("S = AB* AB = A B A = 'a'+ B = 'b'+")
result = insta.parse(parser, "aaaaabbbaaaabb")
print(result)

For a complete working example with result processing, see test/test_instaparse.py which demonstrates how to work with complex pod results including WithMeta objects and transit keywords. Clojure supports metadata on its collections and instaparse results are returned with metadata embeed in them. WithMeta is a simple python class that allows you to work with pods that expect or return metadata along with data.

Cache Management

Pod cache location can be controlled via environment variables:

# Custom cache location
export BABASHKA_PODS_DIR="/custom/path/to/pods"

# Use XDG standard directories
export XDG_CACHE_HOME="/custom/cache"

The resolver automatically handles platform detection and will fall back to compatible architectures when needed (e.g., x86_64 binaries on Apple Silicon with Rosetta).

Patch System

Python Pods includes a runtime patching system that allows you to modify pod behavior at runtime without changing pod code. This enables you to transform pod results, override functions, or add custom data type handling. Most pods naturally assume the pod client is babashka pods, so for some pods, patches make working with them from python more frictionless.

Patch Types

Result Transform Patches - Transform pod function results after execution:

# Transform complex pod results to Python-friendly formats
def unwrap_withmeta(node):
    # Convert WithMeta objects and transit keywords to clean Python data
    # See test/test_instaparse.py for complete implementation
    pass

pods.register_result_transform_patch(
    pod_id, 
    'pod.babashka.instaparse/parse', 
    unwrap_withmeta
)

Code Patches - Replace pod functions entirely with Python code:

python_code = """
result = sum(args[0])  # args available in execution context
"""

pods.register_code_patch(pod_id, 'pod.example/sum-list', python_code)

(for pods with functions that return clj "code")

EDN Reader Patches - Add custom EDN data type handlers:

def read_person(data):
    return Person(data['name'], data['age'])

pods.register_edn_reader_patch(pod_id, 'person', read_person)

Please see the test scripts in test\ for actual code examples.

Example Usage

See test/test_instaparse.py for a complete example where result transform patches automatically clean up complex parse tree results with WithMeta objects and transit keywords.

Patch Management

# List all registered patches
pods.list_patches(pod_id)

# Clear patches for specific pod or all pods
pods.clear_patches(pod_id)  # specific pod
pods.clear_patches()        # all pods

Patches are applied in order: result transforms, then code patches (if present), with code patches taking precedence over pod-provided functionality.

Key Design Choices

Exposing Pod Namespaces as Python Modules

One of the major design decisions in this library is automatically exposing pod namespaces as importable Python modules. When you load a pod:

Each pod namespace becomes a Python module (e.g., pod.test-pod → pod_test_pod)
Pod functions become callable Python functions with proper __doc__ and metadata
Both kebab-case and snake_case naming conventions are supported
Modules are registered in sys.modules for standard Python imports

# After loading a pod, you can import and use it like any Python module
import pod_test_pod as test_pod

# Function names are converted to snake case.
result1 = test_pod.deep_merge(dict1, dict2)  # deep-merge was the function in test-pod

Deferred Namespace Loading

For pods with multiple namespaces, the library supports deferred loading to improve startup performance:

# List available deferred namespaces
pods.list_deferred_namespaces(pod_id)

# Load a deferred namespace on demand
pods.load_and_expose_namespace(pod_id, "pod.example.deferred-ns")

Note pod_id is obtained from when you load a pod. e.g.

pod = pods.load_pod('org.babashka/instaparse', {'version': '0.0.6'})
pod_id = pod['pod_id'] # 'org.babashka/instaparse'

pod_id is used in several of the public api functions which are detailed below.

Data Formats

JSON Support

The library supports standard JSON format for basic data interchange:

# Load a pod with JSON format (default for many pods)
pod = pods.load_pod(["json-pod"])

# JSON automatically handles basic Python types
data = {
    "numbers": [1, 2, 3],
    "text": "hello",
    "boolean": True,
    "nested": {"key": "value"}
}

result = test_pod.process_data(data)

JSON format provides the most basic compatibility and works well for simple data structures. However, it has limitations:

No support for custom types beyond basic JSON types
No metadata preservation
Limited type fidelity (e.g., no distinction between integers and floats in some cases)

For more advanced features like custom types and metadata, consider using Transit+JSON format.

EDN Support

The library supports EDN format with custom readers. To enable custom EDN readers:

# Load pod with custom reader resolution
pod = pods.load_pod(["clojure", "-M:test-pod"], {"resolve": True})

# EDN with custom tags will be automatically converted
# Example: #person {:name "Alice" :age 30} becomes a Python dict with custom structure

Custom EDN readers in pods should follow the standard EDN reader format. The resolve option must be set to True in load_pod() for custom readers to be processed.

Dynamic EDN Handler Registration

You can register custom EDN handlers at runtime:

from edn import TaggedLiteral

# Define a custom type
class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age

# Define read handler
def read_person(data):
    return Person(data['name'], data['age'])

# Define write handler that creates tagged EDN
def write_person(person):
    return TaggedLiteral('myapp/person', {'name': person.name, 'age': person.age})

# Register handlers (must be called within pod context)
pods.add_edn_read_handler('myapp/person', read_person)
pods.add_edn_write_handler(Person, write_person)

# Now Person objects work seamlessly with EDN pods
person = Person("Alice", 30)
result = test_pod.echo(person)  # Preserves Person type

# The write handler creates: #myapp/person {:name "Alice", :age 30}
# The pod parses it, and our read handler converts it back to Person

Transit+JSON Support

For Transit+JSON format, the library uses the transit-python2 library and supports custom read and write transforms:

# Load a pod with Transit+JSON format
pod = pods.load_pod(["clojure", "-M:test-pod", "--transit+json"])

# Custom transforms automatically handle special types
from datetime import datetime
import uuid

# These types are automatically serialized/deserialized
test_datetime = datetime.now()
test_uuid = uuid.uuid4()

# Round-trip through the pod
result_datetime = test_pod.echo(test_datetime)
result_uuid = test_pod.echo(test_uuid)

Built-in Transit Support

The library automatically handles these common types with Transit:

DateTime objects: Serialized with tag "local-date-time" compatible with Java LocalDateTime
UUID objects: Serialized with tag "u" using standard Transit UUID format
Metadata: Special support for preserving metadata on data structures (see below)

Metadata Support with Transit+JSON

Python Pods supports rich metadata preservation using the official Transit "with-meta" tag:

from python_pods import WithMeta

# Create data with metadata
data = [1, 2, 3, 4, 5]
metadata = {"source": "user-input", "timestamp": "2024-01-01", "version": 1}
wrapped_data = WithMeta(data, metadata)

# Send to pod function that preserves metadata
result = test_pod.echo_meta(wrapped_data)

# Check if metadata was preserved
if hasattr(result, 'value') and hasattr(result, 'meta'):
    print(f"Data: {result.value}")
    print(f"Metadata: {result.meta}")
else:
    print("Metadata was not preserved by this pod function")

Note: Metadata preservation depends on the pod function being designed to handle metadata. Functions with arg-meta set to true in their pod definition can receive (and maybe return) WithMeta objects.

Working with Complex Transit Results

When working with pods that return complex transit data structures (like parse trees), you may need to post-process the results to make them more Python-friendly. See test/test_instaparse.py for a complete example of handling WithMeta objects and transit keywords:

def unwrap_withmeta(node):
    """Recursively unwrap WithMeta objects and convert keywords to strings"""
    if hasattr(node, 'value'):
        return unwrap_withmeta(node.value)
    elif isinstance(node, list):
        return [unwrap_withmeta(item) for item in node]
    elif str(type(node)) == "<class 'transit.transit_types.Keyword'>":
        keyword_str = str(node)
        if ' ' in keyword_str:
            name = keyword_str.split(' ')[1].rstrip(' >')
            if '/' in name:
                return name.split('/')[-1]
            return name
        return keyword_str
    else:
        return node

# Convert complex pod results to clean Python data
cleaned_result = unwrap_withmeta(raw_pod_result)

Dynamic Transit Handler Registration

You can register custom Transit handlers at runtime:

# Define custom read handler
class PersonReadHandler:
    @staticmethod
    def from_rep(rep):
        return Person(name=rep["name"], age=rep["age"])

# Define custom write handler  
class PersonWriteHandler:
    @staticmethod
    def tag(obj):
        return "person"
    
    @staticmethod
    def rep(obj):
        return {"name": obj.name, "age": obj.age}

# Register handlers (must be called within pod context)
pods.add_transit_read_handler("person", PersonReadHandler)
pods.add_transit_write_handler([Person], PersonWriteHandler)

# Now Person objects will be automatically serialized/deserialized
person = Person("Alice", 30)
result = test_pod.echo(person)  # Preserves Person type

API Reference

Core Functions

`load_pod(pod_spec, opts=None)`

Load and start a pod process.

Parameters:

pod_spec: Command to run the pod (string or list of strings), or registry pod identifier (e.g., 'org.babashka/instaparse')
opts: Optional configuration dict
- "version": Version to download from registry (required for registry pods)
- "resolve": Enable custom EDN readers (default: False)
- "transport": Use "socket" for socket transport (default: stdio)
- "force": Force re-download from registry (default: False)

Returns: Pod object

Examples:

# Load from registry
pod = pods.load_pod('org.babashka/instaparse', {'version': '0.0.6'})

# Load local pod
pod = pods.load_pod(["clojure", "-M:test-pod"])

# Load with socket transport
pod = pods.load_pod(["my-pod"], {"transport": "socket"})

`unload_pod(pod_id_or_pod)`

Shutdown and cleanup a pod.

Parameters:

pod_id_or_pod: Pod ID string or pod object

`invoke_public(pod_id_or_pod, function_symbol, args, opts=None)`

Directly invoke a pod function without using module imports.

Parameters:

pod_id_or_pod: Pod ID string or pod object
function_symbol: Function name (e.g., "pod.namespace/function-name")
args: List of arguments to pass to the function
opts: Optional configuration dict

Patch System

`register_result_transform_patch(pod_id, function_name, transform_function)`

Register a function to transform pod results after execution.

Parameters:

pod_id: Pod ID string
function_name: Full function name (e.g., "pod.namespace/function")
transform_function: Function that takes result and returns transformed result

Example:

def clean_result(result):
    # Transform complex pod results to Python-friendly format
    return result

pods.register_result_transform_patch(pod_id, 'pod.example/parse', clean_result)

`register_code_patch(pod_id, function_name, python_code)`

Replace a pod function entirely with Python code.

Parameters:

pod_id: Pod ID string
function_name: Full function name (e.g., "pod.namespace/function")
python_code: Python code string (has access to args variable)

Example:

code = "result = sum(args[0])"
pods.register_code_patch(pod_id, 'pod.example/sum-list', code)

`register_edn_reader_patch(pod_id, tag, reader_function)`

Register a custom EDN reader for a specific tag.

Parameters:

pod_id: Pod ID string
tag: EDN tag string (e.g., "person", "date")
reader_function: Function that takes tagged data and returns Python object

`clear_patches(pod_id=None)`

Clear registered patches.

Parameters:

pod_id: Pod ID to clear (optional, clears all if None)

`list_patches(pod_id=None)`

List all registered patches.

Parameters:

pod_id: Pod ID to list (optional, lists all if None)

Module Management

`list_pod_modules()`

List all currently registered pod modules and their functions.

`list_deferred_namespaces(pod_id=None)`

List deferred namespaces for a pod or all pods.

Parameters:

pod_id: Pod ID string (optional, lists all pods if None)

`load_and_expose_namespace(pod_id, namespace_name)`

Load a deferred namespace and expose it as an importable module.

Parameters:

pod_id: Pod ID string
namespace_name: Namespace name to load

Transit Handlers

`add_transit_read_handler(pod_id, tag, handler_class)`

Register a custom Transit read handler for a specific tag.

Parameters:

pod_id: Pod ID string
tag: Transit tag string
handler_class: Class with static from_rep method

Example:

class PersonReadHandler:
    @staticmethod
    def from_rep(rep):
        return Person(rep["name"], rep["age"])

pods.add_transit_read_handler(pod_id, "person", PersonReadHandler)

`add_transit_write_handler(pod_id, classes, handler_class)`

Register a custom Transit write handler for specific classes.

Parameters:

pod_id: Pod ID string
classes: Class or list of classes to handle
handler_class: Class with static tag and rep methods

Example:

class PersonWriteHandler:
    @staticmethod
    def tag(obj):
        return "person"
    
    @staticmethod
    def rep(obj):
        return {"name": obj.name, "age": obj.age}

pods.add_transit_write_handler(pod_id, [Person], PersonWriteHandler)

`set_default_transit_write_handler(pod_id, handler_class)`

Set a default Transit write handler for unregistered types.

Parameters:

pod_id: Pod ID string
handler_class: Class with static tag and rep methods

Data Types

`WithMeta(value, meta=None)`

Container class for data with metadata, used with Transit+JSON format.

Parameters:

value: The actual data value
meta: Metadata dictionary (optional)

Attributes:

value: The wrapped data
meta: The metadata dictionary

Example:

data = [1, 2, 3]
metadata = {"source": "user", "timestamp": "2024-01-01"}
wrapped = pods.WithMeta(data, metadata)

# Send to pod function
result = some_pod_function(wrapped)

# Access result
if hasattr(result, 'meta'):
    print(f"Data: {result.value}")
    print(f"Metadata: {result.meta}")

Exceptions

`PodError(message, data=None)`

Exception raised when pod operations fail.

Attributes:

message: Error message string
data: Additional error data (dict)

Example:

try:
    result = pod_function("invalid_input")
except pods.PodError as e:
    print(f"Pod error: {e}")
    print(f"Error data: {e.data}")

Examples

Basic Usage

import python_pods as pods

# Load a simple pod
pod = pods.load_pod(["echo-pod"])
import pod_echo as echo

result = echo.echo_message("Hello, World!")
print(result)

Working with Complex Data

# Load pod with custom EDN readers
pod = pods.load_pod(["data-pod"], {"resolve": True})
import pod_data as data

# Pod functions can handle complex nested data
nested_data = {
    "users": [
        {"name": "Alice", "scores": [95, 87, 92]},
        {"name": "Bob", "scores": [78, 85, 90]}
    ]
}

processed = data.process_user_data(nested_data)

Custom Transit Handlers

# Load a Transit+JSON pod
pod = pods.load_pod(["my-pod", "--transit+json"])

# Define a custom data type
class Point:
    def __init__(self, x, y):
        self.x = x
        self.y = y
    
    def __eq__(self, other):
        return self.x == other.x and self.y == other.y

# Define handlers
class PointReadHandler:
    @staticmethod
    def from_rep(rep):
        return Point(rep[0], rep[1])

class PointWriteHandler:
    @staticmethod
    def tag(obj):
        return "point"
    
    @staticmethod
    def rep(obj):
        return [obj.x, obj.y]

# Register handlers
pods.add_transit_read_handler("point", PointReadHandler)
pods.add_transit_write_handler([Point], PointWriteHandler)

# Now Point objects work seamlessly with the pod
import pod_my_pod as my_pod

point = Point(10, 20)
result = my_pod.transform_point(point)  # Returns a Point object

Working with Metadata

from transit2 import WithMeta

# Load a pod that supports metadata
pod = pods.load_pod(["metadata-pod", "--transit+json"])
import pod_metadata_pod as meta_pod

# Create data with metadata
data = {"temperature": 23.5, "humidity": 60}
metadata = {
    "sensor_id": "temp_001", 
    "timestamp": "2024-01-01T10:00:00",
    "unit": "celsius"
}

wrapped_data = WithMeta(data, metadata)

# Send to a metadata-aware pod function
result = meta_pod.process_sensor_data(wrapped_data)

# Check if metadata was preserved and enriched
if hasattr(result, 'meta'):
    print(f"Original metadata: {wrapped_data.meta}")
    print(f"Processed metadata: {result.meta}")
    print(f"Processed data: {result.value}")

Async Operations

# Some pods support async operations through callbacks
def handle_result(result):
    print(f"Received: {result}")

def handle_error(error):
    print(f"Error: {error}")

def handle_done():
    print("Operation completed!")

# Use lower-level invoke for async operations
pods.invoke(
    pod, 
    "pod.async/watch-files", 
    ["/path/to/watch"],
    {"handlers": {"success": handle_result, "error": handle_error, "done": handle_done}}
)

Error Handling

The library raises PodError exceptions when pod operations fail:

from python_pods import PodError

try:
    result = test_pod.some_function("invalid_input")
except PodError as e:
    print(f"Pod error: {e}")
    print(f"Error data: {e.data}")

Development and Testing

The project includes a comprehensive test suite using a local test pod. To run tests:

# you've already git cloned this library
# Install dependencies
uv sync

# Run all tests
./run_test.sh

# Or run individual test files
python test/test_instaparse.py

The test pod (in test-pod/) provides example functions for testing various pod features including metadata handling, async operations, and custom data types.

Protocol Compatibility

This library implements the babashka pod protocol and is compatible with any program that implements the pod protocol, regardless of the implementation language. The protocol uses:

Bencode for message framing
EDN, JSON, or Transit+JSON for payload encoding
Standard stdin/stdout or socket communication

License

Distributed under the MIT License.

Name		Name	Last commit message	Last commit date
Latest commit History 12 Commits
.circleci		.circleci
img		img
src		src
test-pod/pod		test-pod/pod
test		test
README.md		README.md
deps.edn		deps.edn
main.py		main.py
pyproject.toml		pyproject.toml
run_tests.sh		run_tests.sh
uv.lock		uv.lock

judepayne/python-pods

Folders and files

Latest commit

History

Repository files navigation

python-pods

Features

Installation

Quick Start

Cache Management

Patch System

Patch Types

Example Usage

Patch Management

Key Design Choices

Exposing Pod Namespaces as Python Modules

Deferred Namespace Loading

Data Formats

JSON Support

EDN Support

Dynamic EDN Handler Registration

Transit+JSON Support

Built-in Transit Support

Metadata Support with Transit+JSON

Working with Complex Transit Results

Dynamic Transit Handler Registration

API Reference

Core Functions

load_pod(pod_spec, opts=None)

unload_pod(pod_id_or_pod)

invoke_public(pod_id_or_pod, function_symbol, args, opts=None)

Patch System

register_result_transform_patch(pod_id, function_name, transform_function)

register_code_patch(pod_id, function_name, python_code)

register_edn_reader_patch(pod_id, tag, reader_function)

clear_patches(pod_id=None)

list_patches(pod_id=None)

Module Management

list_pod_modules()

list_deferred_namespaces(pod_id=None)

load_and_expose_namespace(pod_id, namespace_name)

Transit Handlers

add_transit_read_handler(pod_id, tag, handler_class)

add_transit_write_handler(pod_id, classes, handler_class)

set_default_transit_write_handler(pod_id, handler_class)

Data Types

WithMeta(value, meta=None)

Exceptions

PodError(message, data=None)

Examples

Basic Usage

Working with Complex Data

Custom Transit Handlers

Working with Metadata

Async Operations

Error Handling

Development and Testing

Protocol Compatibility

License

About

Resources

Uh oh!

Stars

Watchers

Forks

Releases

Packages 0

Languages

`load_pod(pod_spec, opts=None)`

`unload_pod(pod_id_or_pod)`

`invoke_public(pod_id_or_pod, function_symbol, args, opts=None)`

`register_result_transform_patch(pod_id, function_name, transform_function)`

`register_code_patch(pod_id, function_name, python_code)`

`register_edn_reader_patch(pod_id, tag, reader_function)`

`clear_patches(pod_id=None)`

`list_patches(pod_id=None)`

`list_pod_modules()`

`list_deferred_namespaces(pod_id=None)`

`load_and_expose_namespace(pod_id, namespace_name)`

`add_transit_read_handler(pod_id, tag, handler_class)`

`add_transit_write_handler(pod_id, classes, handler_class)`

`set_default_transit_write_handler(pod_id, handler_class)`

`WithMeta(value, meta=None)`

`PodError(message, data=None)`

Packages