Fast, simple microservice framework
Project description
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, Topic
from fastmicro.service import Service
class User(BaseModel):
name: str
class Greeting(BaseModel):
name: str
greeting: str
messaging: Messaging = Messaging()
service = Service("test", messaging)
user_topic = Topic[User]("user", messaging)
greeting_topic = Topic[Greeting]("greeting", messaging)
@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()
user = User(name="Greg")
print(user)
greeting = await greet.call(user)
print(greeting)
await service.stop()
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
Run it
$ python hello.py
{'name': 'Greg'}
{'name': 'Greg', 'greeting': 'Hello, Greg!'}
Backends
FastMicro supports the following backends:
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
from pydantic import BaseModel
from fastmicro.messaging.redis import Messaging, Topic
from fastmicro.service import Service
class User(BaseModel):
name: str
class Greeting(BaseModel):
name: str
greeting: str
messaging: Messaging = Messaging()
service = Service("test", messaging)
user_topic = Topic[User]("user", messaging)
greeting_topic = Topic[Greeting]("greeting", messaging)
@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, Topic
from fastmicro.service import Service
class User(BaseModel):
name: str
class Greeting(BaseModel):
name: str
greeting: str
messaging: Messaging = Messaging()
service = Service("test", messaging)
user_topic = Topic[User]("user", messaging)
greeting_topic = Topic[Greeting]("greeting", messaging)
@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 greet.call(user)
print(greeting)
if __name__ == "__main__":
loop = asyncio.get_event_loop()
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.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for fastmicro-0.2.2-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 0e7f8e80a47840b80a077a93a06504896ba5d910bc0fd30f873327802e55b2aa |
|
MD5 | 896fa0b4464127954439f3d6d0796b29 |
|
BLAKE2b-256 | 6f09a0f33daf62d7198bbad8b3bffd6f16d4be04e96bae25dfdd9bdfa6d18d68 |