socketapi 0.2.0

The main goal of SocketAPI is to provide an easy-to-use and flexible framework for building WebSocket APIs using Python. It leverages the power of Starlette for handling WebSocket connections and Pydantic for data validation and serialization. It uses a single multiplexed WebSocket connection, allowing clients to exchange different types of information through endpoint-like actions and channel subscriptions, defined similarly to routes in FastAPI. The framework is inspired by both FastAPI and Phoenix LiveView, combining familiar declarative endpoints with real-time, channel-oriented communication.

Installation

You can install SocketAPI using pip:

pip install socketapi

Simple example

Server

from socketapi import SocketAPI

app = SocketAPI()

# Define "add_numbers" action - endpoint for performing calculations
@app.action("add_numbers")
async def add_numbers(a: int, b: int) -> int:
    return a + b

# Define "chat" channel - subscription for receiving messages
@app.channel("chat")
async def chat_channel(message: str = "Welcome"):
    return {"message": message}

# Action that sends a message to all "chat" channel subscribers
@app.action("send_message")
async def send_message(message: str):
    await chat_channel(message=message)

Run the server with any ASGI server (e.g., Uvicorn):

uvicorn main:app --reload

Client Usage

Connect to the WebSocket endpoint at ws://localhost:8000/ and exchange JSON messages.

Calling an action (request-response pattern)

Send:

{"type": "action", "channel": "add_numbers", "data": {"a": 5, "b": 3}}

Receive:

{"type": "action", "channel": "add_numbers", "status": "completed", "data": 8}

Subscribing to a channel (pub/sub pattern)

Send:

{"type": "subscribe", "channel": "chat"}

Receive confirmation:

{"type": "subscribed", "channel": "chat"}

Broadcasting to channel subscribers

Send:

{"type": "action", "channel": "send_message", "data": {"message": "Hello everyone!"}}

Receive confirmation:

{"type": "action", "channel": "send_message", "status": "completed"}

All subscribers receive:

{"type": "data", "channel": "chat", "data": {"message": "Hello everyone!"}}

How it works:

• Actions (@app.action) - endpoint-like, request-response pattern. Client sends a request and receives a response.
• Channels (@app.channel) - pub/sub pattern. Client subscribes to a channel and automatically receives all data emitted to that channel.
• Single WebSocket - all operations (actions, channels) work through a single WebSocket connection multiplexed via the channel field.

Documentation

The full documentation is available at spaceshaman.github.io/socketapi/

Features and Roadmap

• Define actions with request-response pattern
• Define channels with pub/sub pattern
• Single multiplexed WebSocket connection
• Pydantic models for data validation
• Error handling and validation
• Dependency injection system (like FastAPI)
• Router class for splitting API into multiple files similar to FastAPI
• Broadcasting messages from outside the server context by calling channel-decorated functions
• Error reporting inside decorated handlers with automatic error messages sent to client

Changelog

Changes for each release are thoroughly documented in release notes

Contributing

Contributions are welcome! Feel free to open an issue or submit a pull request. I would like to keep the library to be safe as possible, so i would appreciate if you cover any new feature with tests to maintain 100% coverage.

Install in a development environment

1. First, clone the repository:

   git clone git@github.com:SpaceShaman/socketapi.git
   

2. Install uv if you don't have it:

   curl -LsSf https://astral.sh/uv/install.sh | sh
   

3. Create a virtual environment and install the dependencies:

   cd socketapi
   uv sync
   

Run tests

You can run the tests with the following command:

uv run pytest

You can also run the tests with coverage:

uv run pytest --cov=socketapi

License

This project is licensed under the terms of the MIT license