pydantic-socket 0.1.0

Simple TCP socket client and server with Pydantic models

pydantic_socket - Simple socket client and server with Pydantic models

Examples

Output examples are provided for simultaneous run of Server and Client examples

Server

import asyncio
import logging

import pydantic_socket
from pydantic_socket.types import Action
from pydantic_socket.websocket import (
    Server,
    ServerClient,
)

server = Server()


class TestResponse(pydantic_socket.types.BaseModel):
    foo: int = 1
    bar: str = "bar"
    input: str | None = None


# Register handler for certain action type    
@server.action_handler("hello")
async def hello_handler(s: Server, client: ServerClient, action: Action):
    logging.info(f"Hello, {action.payload.get('name', 'World')}!")
    # Will print:
    # INFO:root:Hello, pydantic_socket!


async def test_handler(s: Server, client: ServerClient, action: Action):
    return TestResponse(foo=1, bar="bar", input=action.payload.get('input'))


# Another way to register handler for certain action type
server.set_action_handler("test", test_handler)

if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    asyncio.run(server.run())

Client

import asyncio
import logging

import pydantic

from pydantic_socket.types import Action
from pydantic_socket.websocket import Client

client = Client("http://127.0.0.1:8080", auto_reconnect=False, client_id=1)


class SomeActionInputPayload(pydantic.BaseModel):
    input: str


async def main():
    # Establishing connection to server
    asyncio.create_task(client.run())
    # Wait some time until connection is done
    await asyncio.sleep(2)
    # Send action
    await client.send(Action(type='hello', payload={"name": "pydantic_socket"}))
    # Send and wait for response
    response = await client.request(Action(type='test', payload={"input": "Some Input"}))

    logging.info(f"Payload is: {response.payload}")
    # Will print: 
    # INFO:root:Payload is: {'foo': 1, 'bar': 'bar', 'input': 'Some Input'}

    await client.close()


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

LICENSE

This project is licensed under the terms of the MIT license.