Propan - just an another one HTTP a declarative Python MQ framework. It's following by fastapi,
simplify Message Brokers around code writing and provides a helpful development toolkit, which existed only in HTTP-frameworks world until now.
It's designed to create reactive microservices around Messaging Architecture.
It is a modern, high-level framework on top of popular specific Python brokers libraries, based on pydantic and fastapi, pytest concepts.
Documentation: https://lancetnik.github.io/Propan/
- Easy: Designed to be easy to use and learn.
- Intuitive: Great editor support. Autocompletion everywhere.
- Dependencies management: Minimize code duplication. Multiple features from each argument and parameter declaration.
- Integrations: Propan is ready to use in pair with any HTTP framework you want
- MQ independent: Single interface to popular MQ:
- RPC: The framework supports RPC requests over MQ, which will allow performing long operations on remote services asynchronously.
- Greate to develop: CLI tool provides great development experience:
- framework-independent way to rule application environment
- application code hot reloading
- Testability: Propan allows you to test your app without external dependencies: you shouldn't suit up a Message Broker, use a virtual one!
async | sync | |
---|---|---|
RabbitMQ | ✔️ stable ✔️ | 🔍 planning 🔍 |
Nats | 🔍 planning 🔍 | |
NatsJS | 🛠️ in progress 🛠️ | 🔍 planning 🔍 |
MQTT | 🔍 planning 🔍 | 🔍 planning 🔍 |
REDIS | 🔍 planning 🔍 | 🔍 planning 🔍 |
Kafka | 🔍 planning 🔍 | 🔍 planning 🔍 |
SQS | 🔍 planning 🔍 | 🔍 planning 🔍 |
If you are interested in this project, please give me feedback by star or/and watch repository.
If you have any questions or ideas about features to implement, welcome to discussions or public telegram group.
With declarative tools you should define what you need to get. With traditional imperative tools you should write what you need to do.
Take a look at classic imperative tools, such as aio-pika, pika, nats-py, etc.
This is the Quickstart with the aio-pika:
import asyncio
import aio_pika
async def main():
connection = await aio_pika.connect_robust(
"amqp://guest:[email protected]/"
)
queue_name = "test_queue"
async with connection:
channel = await connection.channel()
queue = await channel.declare_queue(queue_name)
async with queue.iterator() as queue_iter:
async for message in queue_iter:
async with message.process():
print(message.body)
asyncio.run(main())
aio-pika is a really great tool with a really easy learning curve. But it's still imperative. You need to connect, declare channel, queues, exchanges by yourself. Also, you need to manage connection, message, queue context to avoid any troubles.
It is not a bad way, but it can be easy.
from propan import PropanApp, RabbitBroker
broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
app = PropanApp(broker)
@broker.handle("test_queue")
async def base_handler(body):
print(body)
This is the Propan declarative way to write the same code. That is so much easier, isn't it?
Install using pip
:
$ pip install "propan[async-rabbit]"
# or
$ pip install "propan[async-nats]"
Create an application with the following code at serve.py
:
from propan import PropanApp
from propan import RabbitBroker
# from propan import NatsBroker
broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
# broker = NatsBroker("nats://localhost:4222")
app = PropanApp(broker)
@broker.handle("test")
async def base_handler(body):
'''Handle all default exchange messages with `test` routing key'''
print(body)
And just run it:
$ propan run serve:app
Propan uses pydantic
to cast incoming function arguments to types according to their annotation.
from pydantic import BaseModel
from propan import PropanApp, Context, RabbitBroker
broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
app = PropanApp(broker)
class SimpleMessage(BaseModel):
key: int
@broker.handle("test2")
async def second_handler(body: SimpleMessage):
assert isinstance(body.key, int)
Propan a has dependencies management policy close to pytest fixtures
.
You can specify in functions arguments which dependencies
you would to use. Framework passes them from the global Context object.
Already existed context fields are: app, broker, context (itself), logger and message. If you call not existing field, raises pydantic.ValidationError value.
But you can specify your own dependencies, call dependencies functions (like Fastapi Depends
)
and more.
import aio_pika
from propan import PropanApp, RabbitBroker, Context, Depends
rabbit_broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
app = PropanApp(rabbit_broker)
async def dependency(body: dict) -> bool:
return True
@rabbit_broker.handle("test")
async def base_handler(body: dict,
dep: bool = Depends(dependency),
broker: RabbitBroker = Context()):
assert dep is True
assert broker is rabbit_broker
Propan has its own CLI tool that provided the following features:
- project generation
- multiprocessing workers
- project hot reloading
- custom command line arguments passing
For example: pass your current .env project setting to context
propan run serve:app --env=.env.dev
from propan import PropanApp, RabbitBroker
from propan.annotations import ContextRepo
from pydantic import BaseSettings
broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
app = PropanApp(broker)
class Settings(BaseSettings):
...
@app.on_startup
async def setup(env: str, context: ContextRepo):
settings = Settings(_env_file=env)
context.set_global("settings", settings)
Also, Propan CLI is able to generate a production-ready application template:
propan create [projectname]
Notice: project template require pydantic[dotenv]
installation.
Run the created project:
# Run rabbimq first
docker compose --file [projectname]/docker-compose.yaml up -d
# Run project
propan run [projectname].app.serve:app --env=.env --reload
Now you can enjoy a new development experience!
You can use Propan MQBrokers
without PropanApp
.
Just start and stop them according to your application lifespan.
from propan import NatsBroker
from sanic import Sanic
app = Sanic("MyHelloWorldApp")
broker = NatsBroker("nats://localhost:4222")
@broker.handle("test")
async def base_handler(body):
print(body)
@app.after_server_start
async def start_broker(app, loop):
await broker.start()
@app.after_server_stop
async def stop_broker(app, loop):
await broker.close()
Also, Propan can be used as part of FastAPI.
Just import a PropanRouter you need and declare the message handler
using the @event
decorator. This decorator is similar to the decorator @handle
for the corresponding brokers.
from fastapi import Depends, FastAPI
from pydantic import BaseModel
from propan.fastapi import RabbitRouter
app = FastAPI()
router = RabbitRouter("amqp://guest:guest@localhost:5672")
class Incoming(BaseModel):
m: dict
def call():
return True
@router.event("test")
async def hello(m: Incoming, d = Depends(call)) -> dict:
return { "response": "Hello, Propan!" }
app.include_router(router)
To see more framework usages go to examples/