Metadata-Version: 2.1
Name: uservice
Version: 0.1.2
Summary: A async microservice framework for python
Project-URL: Homepage, https://github.com/dahlkar/uservice
Author-email: Johan Dahlkar <johan.dahlkar@gmail.com>
License-Expression: MIT
Requires-Python: >=3.10
Requires-Dist: aio-pika>=9.0.7
Requires-Dist: click>=8.1.3
Requires-Dist: pydantic>=1.10.8
Requires-Dist: pyyaml>=6.0
Requires-Dist: uvloop>=0.17.0
Requires-Dist: watchfiles>=0.19.0
Provides-Extra: test
Requires-Dist: docker==6.1.3; extra == 'test'
Requires-Dist: pytest-asyncio==0.21.0; extra == 'test'
Requires-Dist: pytest==7.3.1; extra == 'test'
Description-Content-Type: text/markdown

# μService
A async microservice framework for Python. Inspired by [FastAPI](https://github.com/tiangolo/fastapi) and [Nameko](https://github.com/nameko/nameko).
It uses the [uvloop](https://github.com/MagicStack/uvloop) event loop instead of the default python asyncio loop.

## Requirements

Python 3.10+

## Quickstart

Installing using `pip`:

``` shell
$ pip install uservice
```

Create a service `example.py`:

``` python
from uservice import Service


service = Service(name='example')


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

Run the service:

``` shell
$ uservice run example:service
```

## Usage

### Command line

#### `run`

``` shell
$ 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:

``` python
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:

``` python
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`.


``` python
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.

``` python
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](https://fastapi.tiangolo.com/tutorial/dependencies/).

Example dependecy:
``` python
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)
```
