PIfunc

Protocol Interface Functions

One function, every protocol. Everywhere.

Get Started GitHub

What is PIfunc?

PIfunc revolutionizes how you build networked applications by letting you write your function once and expose it via multiple communication protocols simultaneously. No duplicate code. No inconsistencies. Just clean, maintainable, protocol-agnostic code.

Write Once, Run Anywhere

Implement your logic once and expose it via multiple protocols simultaneously. No duplicate code, no inconsistencies.

Zero Boilerplate

Forget about writing boilerplate code for different protocols. Add a single decorator and you're done.

Automatic Protocol Detection

Just specify configuration for the protocols you want to use, and PIfunc automatically activates them. No need to list protocols separately.

Supported Protocols

PIfunc gives you the freedom to use the right protocol for each use case without changing your implementation.

Protocol Description Best For
HTTP/REST RESTful API with JSON Web clients, general API access
gRPC High-performance RPC Microservices, performance-critical systems
MQTT Lightweight pub/sub IoT devices, mobile apps
WebSocket Bidirectional comms Real-time applications, chat
GraphQL Query language Flexible data requirements
ZeroMQ Distributed messaging High-throughput, low-latency systems
AMQP Advanced Message Queuing Enterprise messaging, reliable delivery
Redis In-memory data structure Caching, pub/sub, messaging
CRON Scheduled tasks Periodic jobs, background tasks

Get Started in Minutes

Installation is simple:

pip install pifunc

Create your first multi-protocol function:

from pifunc import service, run_services

@service(
    http={"path": "/api/add", "method": "POST"},
    websocket={"event": "math.add"},
    grpc={}
)
def add(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

if __name__ == "__main__":
    # Protocols are automatically detected from service configurations!
    run_services(
        http={"port": 8080},
        websocket={"port": 8081},
        grpc={"port": 50051},
        watch=True  # Auto-reload on code changes
    )

Run your service:

python your_service.py

Now your 'add' function is simultaneously available through:

Code Examples

Complete Product API
Parameter Handling
Client-Server Pattern
Protocol Filtering
Advanced Config
CLI Usage

Complete Product API Example

This example demonstrates protocol filtering, client-service communication with CRON scheduling, and the API landing page:

# product.py
from random import randint, choice
from string import ascii_letters
import os
import json

# Optional: Filter protocols via environment variable
os.environ["PIFUNC_PROTOCOLS"] = "http,cron"

# Import pifunc after setting environment variables
from pifunc import service, client, run_services


@service(http={"path": "/api/products", "method": "POST"})
def create_product(product: dict) -> dict:
    """Create a new product."""
    return {
        "id": product["id"],
        "name": product["name"],
        "price": product["price"],
        "in_stock": product.get("in_stock", True)
    }


@service(http={"path": "/", "method": "GET"})
def hello() -> dict:
    """API landing page with documentation."""
    return {
        "description": "Create a new product API",
        "path": "/api/products",
        "url": "http://127.0.0.1:8080/api/products/",
        "method": "POST",
        "protocol": "HTTP",
        "version": "1.1",
        "example_data": {
            "id": "1",
            "name": "test",
            "price": "10",
            "in_stock": True
        },
    }


@client(http={"path": "/api/products", "method": "POST"})
@service(cron={"interval": "1m"})
def generate_product() -> dict:
    """Generate a random product every minute."""
    product = {
        "id": str(randint(1000, 9999)),
        "name": ''.join(choice(ascii_letters) for i in range(8)),
        "price": str(randint(10, 100)),
        "in_stock": True
    }
    print(f"Generating random product: {product}")
    return product


if __name__ == "__main__":
    # Protocols are auto-detected, no need to specify them explicitly
    run_services(
        http={"port": 8080},
        cron={"check_interval": 1},
        watch=True
    )

Key Features Demonstrated:

Parameter Handling

@service(
    http={"path": "/api/products", "method": "POST"},
    mqtt={"topic": "products/create"}
)
def create_product(product: dict) -> dict:
    """Create a new product.

    Note: When working with dictionary parameters, use `dict` instead of `Dict`
    for better type handling across protocols.
    """
    return {
        "id": product["id"],
        "name": product["name"],
        "price": product["price"],
        "in_stock": product.get("in_stock", True)
    }

Client-Server Pattern

from pifunc import service, client, run_services
import random

# Server-side service
@service(http={"path": "/api/products", "method": "POST"})
def create_product(product: dict) -> dict:
    """Create a new product."""
    return {
        "id": product["id"],
        "name": product["name"],
        "price": product["price"],
        "created": True
    }

# Client-side function with scheduled execution
@client(http={"path": "/api/products", "method": "POST"})  # Simplified syntax!
@service(cron={"interval": "1h"})  # Run every hour
def generate_product() -> dict:
    """Generate a random product and send it to the create_product service."""
    return {
        "id": f"PROD-{random.randint(1000, 9999)}",
        "name": f"Automated Product {random.randint(1, 100)}",
        "price": round(random.uniform(10.0, 100.0), 2)
    }

if __name__ == "__main__":
    # Protocols are auto-detected from registered services!
    run_services(
        http={"port": 8080},
        cron={"check_interval": 1},
        watch=True
    )

Protocol Filtering with Environment Variables

# Control available protocols via environment variables
import os
os.environ["PIFUNC_PROTOCOLS"] = "http,cron"  # Only enable HTTP and CRON

from pifunc import service, run_services

@service(
    http={"path": "/api/data"},
    grpc={},          # Will be ignored due to PIFUNC_PROTOCOLS
    websocket={}      # Will be ignored due to PIFUNC_PROTOCOLS
)
def get_data():
    return {"status": "success", "data": [...]}

if __name__ == "__main__":
    # Only HTTP and CRON adapters will be loaded
    run_services(
        http={"port": 8080},
        watch=True
    )

Advanced Configuration

@service(
    # HTTP configuration
    http={
        "path": "/api/users/{user_id}",
        "method": "GET",
        "middleware": [auth_middleware, logging_middleware]
    },
    # MQTT configuration
    mqtt={
        "topic": "users/get",
        "qos": 1,
        "retain": False
    },
    # WebSocket configuration
    websocket={
        "event": "user.get",
        "namespace": "/users"
    },
    # GraphQL configuration
    graphql={
        "field_name": "user",
        "description": "Get user by ID"
    }
)
def get_user(user_id: str) -> dict:
    """Get user details by ID."""
    return db.get_user(user_id)

CLI Usage

# Start a service
python your_service.py

# Call a function via HTTP (default protocol)
pifunc call add --args '{"a": 5, "b": 3}'

# Call a function with specific protocol
pifunc call add --protocol grpc --args '{"a": 5, "b": 3}'

# Generate client code
pifunc generate client --language python --output client.py

# View service documentation
pifunc docs serve
Full Documentation More Examples

Why Developers Love PIfunc

DRY Code

Don't Repeat Yourself. Implement business logic once, not for each protocol.

Protocol Auto-Detection

Just specify configuration for each protocol you want to use. No need to maintain redundant protocol lists.

Environment Control

Limit available protocols with PIFUNC_PROTOCOLS environment variable for different deployment environments.

Simplified Client Syntax

Create client functions with clean syntax that automatically detects which protocol to use.

Hot Reload

Change code and see updates immediately without restarting servers.

Production Ready

Built for real-world applications with monitoring and scaling in mind.

Features

Testing

# Install development dependencies
pip install -r requirements-dev.txt

# Run all tests
pytest

# Run specific test categories
pytest tests/test_http_adapter.py
pytest tests/test_integration.py

Contributing

Contributions are welcome! See CONTRIBUTING.md for how to get started.

  1. Fork the repository
  2. Create your feature branch (git checkout -b feature/amazing-feature)
  3. Commit your changes (git commit -m 'Add some amazing feature')
  4. Push to the branch (git push origin feature/amazing-feature)
  5. Open a Pull Request