mediatr


Namemediatr JSON
Version 1.3.2 PyPI version JSON
download
home_pagehttps://github.com/megafetis/mediatr_py
Summarymediator and CQRS pattern implementation with pipline behaviors for Python 3.6+. Mediatr py
upload_time2021-04-20 12:38:08
maintainer
docs_urlNone
authorEvgeniy Fetisov
requires_python>=3.6
licenseMIT -or- Apache License 2.0
keywords mediator mediatr cqrs cqrs mediatr_py mediator py mediatr py pipline behaviors command query responsability segregation command bus bus
VCS
bugtrack_url
requirements No requirements were recorded.
Travis-CI No Travis.
coveralls test coverage No coveralls.
            # mediatr_py

[![PyPI](https://img.shields.io/pypi/v/mediatr)](https://pypi.org/project/mediatr)
[![Python](https://img.shields.io/pypi/pyversions/mediatr)](https://pypi.org/project/mediatr) 
[![Downloads](https://img.shields.io/pypi/dm/mediatr)](https://pypi.org/project/mediatr) 

<a href="https://www.buymeacoffee.com/megafetis" target="_blank"><img src="https://www.buymeacoffee.com/assets/img/custom_images/orange_img.png" alt="Buy Me A Coffee" style="height: 41px !important;width: 174px !important;box-shadow: 0px 3px 2px 0px rgba(190, 190, 190, 0.5) !important;-webkit-box-shadow: 0px 3px 2px 0px rgba(190, 190, 190, 0.5) !important;" ></a>

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

It is a port of [Mediatr](https://github.com/jbogard/MediatR) from .Net C#

Requirements:
* Python >= 3.6

## Usage:

install [mediatr](https://pypi.org/project/mediatr/):

`pip install mediatr`

### Define your request class

```py

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

```

### Define your handler class or function

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

```py
@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

```py
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]

```

> If you are using synchronous `mediator.send(request)` method, try to define synchronous handlers and behaviors
>
> In another case use `asyncio` module for manual manage of event loop in synchronous code


### Run mediator statically, without instance

```py
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 not 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:

```py
@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'

```

## Using custom handler (behavior) factory for handlers (behaviors) as classes

If your handlers or behaviors registered as functions, it just executes them.

In case with handlers or behaviors, declared as classes with method `handle` Mediator uses function, that instantiates handlers or behaviors:

```py
def default_handler_class_manager(HandlerCls:type,is_behavior:bool=False):
    return HandlerCls()

```


For example, if you want to instantiate them with dependency injector or custom, pass your own factory function to Mediator:

```py
def my_class_handler_manager(handler_class, is_behavior=False):
    
    if is_behavior:
        # custom logic
        pass

    return injector.get(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.

## Using with generic typing support (version >= 1.2):

```py

from mediatr import Mediator, GenericQuery


class UserModel(BaseModel): # For example sqlalchemy ORM entity
    id = Column(String,primary_key=True)
    name = Column(String)


class FetchUserQuery(GenericQuery[UserModel])
    def __init__(self,user_id:str):
        self.user_id = user_id


mediator = Mediator()

request = FetchUserQuery(user_id = "123456")

user = mediator.send(request) # type of response will be a UserModel


# -------------------------------------------------------------


class FetchUserQueryHandler():

    def handle(self, request:FetchUserQuery):
        db_session = Session() #sqlalchemy session
        return db_session.query(UserModel).filter(UserModel.id == request.user_id).one()

# or handler as simple function:

def fetch_user_query_handler(request:FetchUserQuery):
    db_session = Session() #sqlalchemy session
    return db_session.query(UserModel).filter(UserModel.id == request.user_id).one()


```

Please give a star if the library is useful for you :smiley:
            

Raw data

            {
    "_id": null,
    "home_page": "https://github.com/megafetis/mediatr_py",
    "name": "mediatr",
    "maintainer": "",
    "docs_url": null,
    "requires_python": ">=3.6",
    "maintainer_email": "",
    "keywords": "mediator,mediatr,CQRS,cqrs,mediatr_py,mediator py,mediatr py,pipline,behaviors,command,query,responsability,segregation,command bus,bus",
    "author": "Evgeniy Fetisov",
    "author_email": "me@efetisov.ru",
    "download_url": "https://files.pythonhosted.org/packages/a0/0f/ddba08ed784ced1fce3ac895151b377d71f768ee70214f4cff479a754f58/mediatr-1.3.2.tar.gz",
    "platform": "",
    "description": "# mediatr_py\n\n[![PyPI](https://img.shields.io/pypi/v/mediatr)](https://pypi.org/project/mediatr)\n[![Python](https://img.shields.io/pypi/pyversions/mediatr)](https://pypi.org/project/mediatr) \n[![Downloads](https://img.shields.io/pypi/dm/mediatr)](https://pypi.org/project/mediatr) \n\n<a href=\"https://www.buymeacoffee.com/megafetis\" target=\"_blank\"><img src=\"https://www.buymeacoffee.com/assets/img/custom_images/orange_img.png\" alt=\"Buy Me A Coffee\" style=\"height: 41px !important;width: 174px !important;box-shadow: 0px 3px 2px 0px rgba(190, 190, 190, 0.5) !important;-webkit-box-shadow: 0px 3px 2px 0px rgba(190, 190, 190, 0.5) !important;\" ></a>\n\nThis is an async implementation of Mediator pattern with pipline behaviors.\n\nIt is a port of [Mediatr](https://github.com/jbogard/MediatR) from .Net C#\n\nRequirements:\n* Python >= 3.6\n\n## Usage:\n\ninstall [mediatr](https://pypi.org/project/mediatr/):\n\n`pip install mediatr`\n\n### Define your request class\n\n```py\n\nclass GetArrayQuery():\n    def __init__(self,items_count:int):\n        self.items_count = items_count\n\n```\n\n### Define your handler class or function\n\n```py\nimport Mediator from mediatr\n\n@Mediator.handler\nasync def get_array_handler(request:GetArrayQuery):\n    items = list()\n    for i in range(0, request.items_count):\n        items.append(i)\n    return items\n    \n# or just Mediator.register_handler(get_array_handler)\n    \n```\n\nor class:\n\n```py\n@Mediator.handler\nclass GetArrayQueryHandler():\n    def handle(self,request:GetArrayQuery):\n        items = list()\n        for i in range(0, request.items_count):\n            items.append(i)\n        return items\n        \n# or just Mediator.register_handler(GetArrayQueryHandler)\n```\n\n### Run mediator\n\n```py\nimport Mediator from mediatr\n\nmediator = Mediator()\n\nrequest = GetArrayQuery(5)\n\nresult = await mediator.send_async(request)\n\n# result = mediator.send(request) in synchronous mode\n\nprint(result) // [0,1,2,3,4]\n\n```\n\n> If you are using synchronous `mediator.send(request)` method, try to define synchronous handlers and behaviors\n>\n> In another case use `asyncio` module for manual manage of event loop in synchronous code\n\n\n### Run mediator statically, without instance\n\n```py\nimport Mediator from mediatr\n\nrequest = GetArrayQuery(5)\n\nresult = await Mediator.send_async(request)\n# or:\nresult = Mediator.send(request) #in synchronous mode. Async handlers and behaviors will not blocking!\n\nprint(result) // [0,1,2,3,4]\n\n```\n\nNote that instantiation of `Mediator(handler_class_manager = my_manager_func)` is useful if you have custom handlers creation. For example using an injector.\nBy default class handlers are instantiated with simple init:  `SomeRequestHandler()`. handlers or behaviors as functions are executed directly. \n\n\n## Using behaviors\nYou can define behavior class with method 'handle' or function:\n\n```py\n@Mediator.behavior\nasync def get_array_query_behavior(request:GetArrayQuery, next): #behavior only for GetArrayQuery or derived classes\n    array1 = await next()\n    array1.append(5)\n    return array1\n\n@Mediator.behavior\ndef common_behavior(request:object, next): #behavior for all requests because issubclass(GetArrayQuery,object)==True\n    request.timestamp = '123'\n    return next()\n\n# ...\n\nmediator = Mediator()\nrequest = GetArrayQuery(5)\nresult = await mediator.send_async(request)\nprint(result) // [0,1,2,3,4,5]\nprint(request.timestamp) // '123'\n\n```\n\n## Using custom handler (behavior) factory for handlers (behaviors) as classes\n\nIf your handlers or behaviors registered as functions, it just executes them.\n\nIn case with handlers or behaviors, declared as classes with method `handle` Mediator uses function, that instantiates handlers or behaviors:\n\n```py\ndef default_handler_class_manager(HandlerCls:type,is_behavior:bool=False):\n    return HandlerCls()\n\n```\n\n\nFor example, if you want to instantiate them with dependency injector or custom, pass your own factory function to Mediator:\n\n```py\ndef my_class_handler_manager(handler_class, is_behavior=False):\n    \n    if is_behavior:\n        # custom logic\n        pass\n\n    return injector.get(handler_class)\n\nmediator = Mediator(handler_class_manager=my_class_handler_manager)\n\n```\nPS:\n\n\nThe `next` function in behavior is `async`, so if you want to take results or if your behavior is async, use `middle_results = await next()`\n\n\nHandler may be async too, if you need.\n\n## Using with generic typing support (version >= 1.2):\n\n```py\n\nfrom mediatr import Mediator, GenericQuery\n\n\nclass UserModel(BaseModel): # For example sqlalchemy ORM entity\n    id = Column(String,primary_key=True)\n    name = Column(String)\n\n\nclass FetchUserQuery(GenericQuery[UserModel])\n    def __init__(self,user_id:str):\n        self.user_id = user_id\n\n\nmediator = Mediator()\n\nrequest = FetchUserQuery(user_id = \"123456\")\n\nuser = mediator.send(request) # type of response will be a UserModel\n\n\n# -------------------------------------------------------------\n\n\nclass FetchUserQueryHandler():\n\n    def handle(self, request:FetchUserQuery):\n        db_session = Session() #sqlalchemy session\n        return db_session.query(UserModel).filter(UserModel.id == request.user_id).one()\n\n# or handler as simple function:\n\ndef fetch_user_query_handler(request:FetchUserQuery):\n    db_session = Session() #sqlalchemy session\n    return db_session.query(UserModel).filter(UserModel.id == request.user_id).one()\n\n\n```\n\nPlease give a star if the library is useful for you :smiley:",
    "bugtrack_url": null,
    "license": "MIT -or- Apache License 2.0",
    "summary": "mediator and CQRS pattern implementation with pipline behaviors for Python 3.6+. Mediatr py",
    "version": "1.3.2",
    "split_keywords": [
        "mediator",
        "mediatr",
        "cqrs",
        "cqrs",
        "mediatr_py",
        "mediator py",
        "mediatr py",
        "pipline",
        "behaviors",
        "command",
        "query",
        "responsability",
        "segregation",
        "command bus",
        "bus"
    ],
    "urls": [
        {
            "comment_text": "",
            "digests": {
                "md5": "d8522b584876e78e3edf0b05c2e813d9",
                "sha256": "6a463c53fe35982cfff3e273a661cf317c46a850d3f0385493c0d7571c06b89f"
            },
            "downloads": -1,
            "filename": "mediatr-1.3.2.tar.gz",
            "has_sig": false,
            "md5_digest": "d8522b584876e78e3edf0b05c2e813d9",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": ">=3.6",
            "size": 10402,
            "upload_time": "2021-04-20T12:38:08",
            "upload_time_iso_8601": "2021-04-20T12:38:08.042726Z",
            "url": "https://files.pythonhosted.org/packages/a0/0f/ddba08ed784ced1fce3ac895151b377d71f768ee70214f4cff479a754f58/mediatr-1.3.2.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2021-04-20 12:38:08",
    "github": true,
    "gitlab": false,
    "bitbucket": false,
    "github_user": null,
    "github_project": "megafetis",
    "error": "Could not fetch GitHub repository",
    "lcname": "mediatr"
}
        
Elapsed time: 0.26014s