unitpulse-feign 0.2.1

Declarative HTTP client SDK for Python microservices

unitpulse-feign

Declarative HTTP client for Python microservices, inspired by Spring Cloud OpenFeign. Integrates with unitpulse-discovery to call services by name with automatic load balancing.

Installation

pip install "./unitpulse-feign"

Usage

from unitpulse_feign import setup_feign, get, post, put, delete, patch

# Must be called after setup_discovery
setup_feign(app)

@app.feign.client("target-service", lb_strategy="round_robin")
class TargetClient:
    @get("/api/users/{user_id}")
    def get_user(self, user_id: int) -> dict: pass

    @post("/api/users")
    def create_user(self, body: dict) -> dict: pass

# Create singleton and reuse — do NOT instantiate per request
client = TargetClient()
response = client.get_user(user_id=123)
response = client.create_user(body={"name": "Alice"})

Parameters

setup_feign(app)

Attaches app.feign to the app. Must be called after setup_discovery.

@app.feign.client(...)

Parameter	Type	Default	Description
service_name	str	—	Target service name (discovery mode)
host	str	—	Direct host (direct mode, no discovery)
port	int	—	Direct port (direct mode)
scheme	str	"http"	"http" or "https"
timeout	int	5	Request timeout (seconds)
lb_strategy	str	"round_robin"	Load balancing strategy

Modes:

• Discovery mode: provide service_name — instances resolved via app.discovery
• Direct mode: provide host + port — bypasses discovery

HTTP Method Decorators

Decorator	Method
@get(path)	GET
@post(path)	POST
@put(path)	PUT
@delete(path)	DELETE
@patch(path)	PATCH

Path supports {variable} placeholders matched to keyword arguments:

@get("/api/users/{user_id}/orders/{order_id}")
def get_order(self, user_id: int, order_id: int) -> dict: pass

client.get_order(user_id=1, order_id=42)

Load Balancing Strategies

Strategy	Description
round_robin	Cycles through instances in order
random	Randomly selects an instance
weighted	Weighted random based on weight in instance metadata
least_conn	Selects instance with fewest active connections

Async Support

Each method exposes .async_call for use in async contexts:

result = await client.get_user.async_call(user_id=123)

License

MIT