python-pods 0.1.0

A faithful port of babashka pods to python

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
• Pod functionality patching system via pyproject.toml configuration
• 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.

# Install from Pypi
uv add python-pods
# Install dependencies and activate environment
uv sync
source .venv/bin/activate

# Run tests
./run_test.sh

Quick Start

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. 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.

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)

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:

1. Each pod namespace becomes a Python module (e.g., pod.test-pod → pod_test_pod)
2. Pod functions become callable Python functions with proper __doc__ and metadata
3. Both kebab-case and snake_case naming conventions are supported
4. 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['id']

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:

# 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

Copyright © 2025 Jude Payne

Distributed under the MIT License.