uservice 0.1.2

A async microservice framework for python

μService

A async microservice framework for Python. Inspired by FastAPI and Nameko. It uses the uvloop event loop instead of the default python asyncio loop.

Requirements

Python 3.10+

Quickstart

Installing using pip:

$ pip install uservice

Create a service example.py:

from uservice import Service


service = Service(name='example')


@service.event_handler('source', 'event')
def handle_event(payload):
    print(payload)

Run the service:

$ uservice run example:service

Usage

Command line

run

$ uservice run --help
Usage: uservice run [OPTIONS] SERVICE

Options:
  --reload           Enable auto-reload.
  --workers INTEGER  Number of worker processes. Not valid with --reload
                     [default: 1]
  --help             Show this message and exit.

Events (Pub-Sub)

At the moment only amqp is supported for events.

Subscribe

Event handlers in uservice require that the kwarg payload is in the function that handles it. Event handlers in uservice support runtime typechecking using pydantic, example:

from uservice import Service
from pydantic import BaseModel


service = Service(name='service')


class Payload(BaseModel):
    foo: int
    bar: int
    

@service.event_handler('source', 'event')
def handle_event(payload: Payload):
    print(payload)

If an event is sent to this event_handler not matching the Payload schema it will raise a ValidationError.

Publish

Event publishing in uservice is handled as a dependency injection. It aslo supports validation of payloads using pydantic, example:

from uservice import Service, EventPublisher, Depends
from typing import Annotated
from pydantic import BaseModel


class Payload(BaseModel):
    foo: int
    bar: str


publisher = EventPublisher(publish_model=Payload)

service = Service(name="service")

@service.event_handler("source", "event")
def handle(
        payload: Payload,
        publish: Annotated[Callable, Depends(publisher)],
):
    publish("event", payload)

RPC

At the moment rpcs only support amqp. Rpcs in uservice supports validation of reponses using pydantic.

from uservice import Service
from pydantic import BaseModel


class Response(BaseModel):
    foo: int
    bar: str


service = Service(name="service")

@service.rpc(response_model=Response)
def method(x: int, b: int):
    return {
        'foo': x,
        'bar': y,
    }

Example of calling a rpc from another service. When calling a rpc only support kwargs.

from uservice import Service, RpcProxy, ServiceProxy, Depends
from typing import Annotated
from pydantic import BaseModel


caller = Service(name="caller")

@service.rpc()
def call(service: Annotated[ServiceProxy, Depends(RpcProxy)]):
    print(service.method(x=2, y=3))

Dependency Injection

uservice uses a dependency injection system which is heavily inspired by FastAPI.

Example dependecy:

from uservice import Service, Depends
from typing import Annotated
from pydantic import BaseModel, BaseSettings


class Payload(BaseModel):
    foo: int
    bar: str


service = Service(name="service")

class Settings(BaseSettings):
    version: str = "1.0.0"


def get_settings():
    return Settings()
        

@service.event_handler("source", "event")
def handle(
        payload: Payload,
        settings: Annotated[Settings, Depends(get_settings)],
):
    print(settings.version, payload)