Skip to main content

mediator and CQRS pattern implementation with pipline behaviors for Python 3.5+. Mediatr py

Project description

mediatr_py

This is an async implementation of Mediator pattern with pipline behaviors.

It is a port of Mediatr from .Net C#

Requirements:

  • Python >= 3.5

Usage:

install mediatr:

pip install mediatr

Define your request class

class GetArrayQuery():
    def __init__(self,items_count:int):
        self.items_count = items_count

Define your handler class or function

import Mediator from mediatr

@Mediator.handler
async def get_array_handler(request:GetArrayQuery):
    items = list()
    for i in range(0, request.items_count):
        items.append(i)
    return items
    
# or just Mediator.register_handler(get_array_handler)
    

or class:

@Mediator.handler
class GetArrayQueryHandler():
    def handle(self,request:GetArrayQuery):
        items = list()
        for i in range(0, request.items_count):
            items.append(i)
        return items
        
# or just Mediator.register_handler(GetArrayQueryHandler)

Run mediator

import Mediator from mediatr

mediator = Mediator()

request = GetArrayQuery(5)

result = await mediator.send_async(request)

# result = mediator.send(request) in synchronous mode

print(result) // [0,1,2,3,4]

Run mediator statically, without instance

import Mediator from mediatr

request = GetArrayQuery(5)

result = await Mediator.send_async(request)
# or:
result = Mediator.send(request) #in synchronous mode. Async handlers and behaviors will executed with blocking

print(result) // [0,1,2,3,4]

Note that instantiation of Mediator(handler_class_manager = my_manager_func) is useful if you have custom handlers creation. For example using an injector. By default class handlers are instantiated with simple init: SomeRequestHandler(). handlers or behaviors as functions are executed directly.

Using behaviors

You can define behavior class with method 'handle' or function:

@Mediator.behavior
async def get_array_query_behavior(request:GetArrayQuery, next): #behavior only for GetArrayQuery or derived classes
    array1 = await next()
    array1.append(5)
    return array1

@Mediator.behavior
def common_behavior(request:object, next): #behavior for all requests because issubclass(GetArrayQuery,object)==True
    request.timestamp = '123'
    return next()

# ...

mediator = Mediator()
request = GetArrayQuery(5)
result = await mediator.send_async(request)
print(result) // [0,1,2,3,4,5]
print(request.timestamp) // '123'

Create yor own class handler manager function

For example, if you want to instantiate them with dependency injector or custom

def my_class_handler_manager(handler_class, is_behavior=False):
    return handler_class()

mediator = Mediator(handler_class_manager=my_class_handler_manager)

PS:

The 'next' function in behavior is async, so if you want to take results or if your behavior is async, use middle_results = await next()

Handler may be async too, if you need.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for mediatr, version 1.1.2.2
Filename, size File type Python version Upload date Hashes
Filename, size mediatr-1.1.2.2.tar.gz (4.7 kB) File type Source Python version None Upload date Hashes View

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring DigiCert DigiCert EV certificate Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page