FastMicro

Fast, simple microservice framework

---

FastMicro is a modern, fast (high-performance) framework for building microservices with Python 3.7+ based on asyncio.

Install

To install FastMicro run the following:

$ pip install fastmicro[redis]

Example

This example shows how to use the default in-memory backend for evaluation and testing.

Note:

The in-memory backend cannot be used for inter-process communication.

Create it

• Create a file hello.py with:

#!/usr/bin/env python3

import asyncio
from pydantic import BaseModel

from fastmicro.messaging.memory import Messaging
from fastmicro.service import Service


class User(BaseModel):
    name: str


class Greeting(BaseModel):
    name: str
    greeting: str


service = Service("test")
loop = asyncio.get_event_loop()
messaging = Messaging(loop=loop)
user_topic = messaging.topic("user", User)
greeting_topic = messaging.topic("greeting", Greeting)


@service.entrypoint(user_topic, greeting_topic)
async def greet(user: User) -> Greeting:
    greeting = Greeting(name=user.name, greeting=f"Hello, {user.name}!")
    return greeting


async def main() -> None:
    await service.start()

    async with messaging:
        user = User(name="Greg")
        print(user)
        greeting = await service.greet(user)
        print(greeting)

    await service.stop()


if __name__ == "__main__":
    loop.run_until_complete(main())

Run it

$ python hello.py
{'name': 'Greg'}
{'name': 'Greg', 'greeting': 'Hello, Greg!'}

Backends

FastMicro supports the following backends:

• Kafka
• Redis

To install FastMicro with one of these backends run one of the following:

$ pip install fastmicro[kafka]
$ pip install fastmicro[redis]

Another Example

This example shows how to use the Redis backend for inter-process communication.

Create it

• Create a file example.py with:

#!/usr/bin/env python3

import asyncio
from pydantic import BaseModel

from fastmicro.messaging.redis import Messaging
from fastmicro.service import Service


class User(BaseModel):
    name: str


class Greeting(BaseModel):
    name: str
    greeting: str


service = Service("test")
loop = asyncio.get_event_loop()
messaging = Messaging(loop=loop)
user_topic = messaging.topic("user", User)
greeting_topic = messaging.topic("greeting", Greeting)


@service.entrypoint(user_topic, greeting_topic)
async def greet(user: User) -> Greeting:
    print(user)
    greeting = Greeting(name=user.name, greeting=f"Hello, {user.name}!")
    print(greeting)
    return greeting


if __name__ == "__main__":
    service.run()

• Create a file test.py with:

#!/usr/bin/env python3

import asyncio
from pydantic import BaseModel

from fastmicro.messaging.redis import Messaging
from fastmicro.service import Service


class User(BaseModel):
    name: str


class Greeting(BaseModel):
    name: str
    greeting: str


service = Service("test")
loop = asyncio.get_event_loop()
messaging = Messaging(loop=loop)
user_topic = messaging.topic("user", User)
greeting_topic = messaging.topic("greeting", Greeting)


@service.entrypoint(user_topic, greeting_topic)
async def greet(user: User) -> Greeting:
    ...


async def main() -> None:
    async with messaging:
        user = User(name="Greg")
        print(user)
        greeting = await service.greet(user)
        print(greeting)


if __name__ == "__main__":
    loop.run_until_complete(main())

Run it

• In a terminal run:

$ python example.py
{'name': 'Greg'}
{'name': 'Greg', 'greeting': 'Hello, Greg!'}
^C

• In another terminal run:

$ python test.py
{'name': 'Greg'}
{'name': 'Greg', 'greeting': 'Hello, Greg!'}

License

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