slimfaas-client
Python client to connect Jobs or virtual functions to SlimFaas via WebSocket. Lets any process receive async requests and publish/subscribe events without exposing an HTTP port.
Links:
- PyPI package: https://pypi.org/project/slimfaas-client
- GitHub repository: https://github.com/SlimPlanet/SlimFaas
- SlimFaas website: https://slimfaas.dev/
Requirements
- Python ≥ 3.10
- UV as package manager
Installation
uv add slimfaas-client
# or from source
uv pip install -e .
Quick start
import asyncio
from slimfaas_client import (
SlimFaasClient, SlimFaasClientConfig,
SubscribeEventConfig, FunctionVisibility,
AsyncRequest, PublishEvent,
)
async def handle_request(req: AsyncRequest) -> int:
"""Called when SlimFaas sends an async-function request.
Return an HTTP status code (200 = success, 500 = error, 202 = long processing)."""
print(f"{req.method} {req.path}{req.query}")
print(f"Body: {req.body}")
return 200
async def handle_event(evt: PublishEvent) -> None:
"""Called when SlimFaas publishes a publish-event."""
print(f"Event '{evt.event_name}': {evt.body}")
async def main():
config = SlimFaasClientConfig(
function_name="my-job",
subscribe_events=[
SubscribeEventConfig(name="order-created"),
SubscribeEventConfig(name="order-updated"),
],
default_visibility=FunctionVisibility.PUBLIC,
number_parallel_request=5,
)
async with SlimFaasClient("ws://slimfaas:5003/ws", config) as client:
client.on_async_request(handle_request)
client.on_publish_event(handle_event)
await client.run_forever()
asyncio.run(main())
Full configuration
from slimfaas_client import (
SlimFaasClientConfig, SubscribeEventConfig, PathVisibilityConfig,
FunctionVisibility, FunctionTrust,
)
config = SlimFaasClientConfig(
function_name="my-job",
# SlimFaas/DependsOn
depends_on=["other-function"],
# SlimFaas/SubscribeEvents — each entry may override visibility individually
subscribe_events=[
SubscribeEventConfig(name="my-event", visibility=FunctionVisibility.PUBLIC),
SubscribeEventConfig(name="internal-event"), # inherits default_visibility
],
# SlimFaas/DefaultVisibility
default_visibility=FunctionVisibility.PUBLIC, # or PRIVATE
# SlimFaas/PathsStartWithVisibility
paths_start_with_visibility=[
PathVisibilityConfig(path="/admin", visibility=FunctionVisibility.PRIVATE),
],
# SlimFaas/Configuration
configuration='{"key": "value"}',
# SlimFaas/ReplicasStartAsSoonAsOneFunctionRetrieveARequest
replicas_start_as_soon_as_one_function_retrieve_a_request=True,
# SlimFaas/NumberParallelRequest
number_parallel_request=10,
# SlimFaas/NumberParallelRequestPerPod
number_parallel_request_per_pod=5,
# SlimFaas/DefaultTrust
default_trust=FunctionTrust.TRUSTED, # or UNTRUSTED
)
Sync streaming (HTTP-over-WebSocket)
from slimfaas_client import SyncRequest
async def handle_sync(req: SyncRequest) -> None:
body = b'{"status": "ok"}'
await req.response.start(200, {"Content-Type": ["application/json"]})
await req.response.write(body)
await req.response.complete()
client.on_sync_request(handle_sync)
Long-running requests (status 202)
Return 202 to acknowledge the request without completing it yet,
then call send_callback when done:
async def handle_long(req: AsyncRequest) -> int:
asyncio.create_task(process_in_background(req))
return 202 # "I'll handle it — will call back"
async def process_in_background(req: AsyncRequest) -> None:
await asyncio.sleep(10)
await client.send_callback(req.element_id, 200)
Dependency injection
The handlers are plain async functions, so you can close over any dependency you resolved from your DI framework:
# Example with a database session from SQLAlchemy
from sqlalchemy.ext.asyncio import AsyncSession
async def make_handler(session: AsyncSession):
async def handle_request(req: AsyncRequest) -> int:
await session.execute(...) # use the injected session
return 200
return handle_request
client.on_async_request(await make_handler(db_session))
Automatic reconnection
The client reconnects automatically after a disconnection. Configure the delay between attempts and the keepalive ping interval:
client = SlimFaasClient(
"ws://...",
config,
reconnect_delay=10.0,
ping_interval=30.0, # use 0 to disable keepalive pings
)
Important rules
-
function_namemust not match an existing Kubernetes Deployment name. SlimFaas will reject the registration with aSlimFaasRegistrationError. -
All clients sharing the same
function_namemust have the exact same configuration. Mismatches are rejected on connection.
Development
uv sync --extra dev
uv run pytest
1 maintainer
Release files for slimfaas-client 0.84.14
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| slimfaas_client-0.84.14.tar.gz | 79.7 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| slimfaas_client-0.84.14-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 92.9 kB
Release files / slimfaas_client-0.84.14.tar.gz
| Download URL | slimfaas_client-0.84.14.tar.gz |
|---|---|
| Size | 79.7 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
a5ef089b33f41c71514f57995612ae59af5bf395acc3e1eb384742248cd0816d
|
|
BLAKE2b-256 checksum How to use checksums |
3734a981b944f5c9f06a83c953285202dd07405388495c9133eee4e39defb812
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Release files / slimfaas_client-0.84.14-py3-none-any.whl
| Download URL | slimfaas_client-0.84.14-py3-none-any.whl |
|---|---|
| Size | 13.2 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
9d7728c0a27b69f869b1cfb0336f6ad00ea98a9f74f5d8c115e2883d34068503
|
|
BLAKE2b-256 checksum How to use checksums |
cb51cf0e34f311e1f6979dea7c0f4d0e5feb6417d2c36be7bdef23e870f8e458
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
