Prodream/ormar
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bump version, more tests, update docs

Browse Source
This commit is contained in:
collerek
2020-12-06 19:45:09 +01:00
parent 420826f472
commit 9f86e1d46e
10 changed files with 419 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
ormar/decorators/signals.py
View File

@ -1,11 +1,11 @@
from typing import Any, Callable, List, TYPE_CHECKING, Type, Union
from typing import Callable, List, TYPE_CHECKING, Type, Union
if TYPE_CHECKING: # pragma: no cover
from ormar import Model
def receiver(
signal: str, senders: Union[Type["Model"], List[Type["Model"]]], **kwargs: Any
signal: str, senders: Union[Type["Model"], List[Type["Model"]]]
) -> Callable:
def _decorator(func: Callable) -> Callable:
if not isinstance(senders, list):
@ -14,43 +14,31 @@ def receiver(
_senders = senders
for sender in _senders:
signals = getattr(sender.Meta.signals, signal)
signals.connect(func, **kwargs)
signals.connect(func)
return func
return _decorator
def post_save(
senders: Union[Type["Model"], List[Type["Model"]]], **kwargs: Any
) -> Callable:
return receiver(signal="post_save", senders=senders, **kwargs)
def post_save(senders: Union[Type["Model"], List[Type["Model"]]],) -> Callable:
return receiver(signal="post_save", senders=senders)
def post_update(
senders: Union[Type["Model"], List[Type["Model"]]], **kwargs: Any
) -> Callable:
return receiver(signal="post_update", senders=senders, **kwargs)
def post_update(senders: Union[Type["Model"], List[Type["Model"]]],) -> Callable:
return receiver(signal="post_update", senders=senders)
def post_delete(
senders: Union[Type["Model"], List[Type["Model"]]], **kwargs: Any
) -> Callable:
return receiver(signal="post_delete", senders=senders, **kwargs)
def post_delete(senders: Union[Type["Model"], List[Type["Model"]]],) -> Callable:
return receiver(signal="post_delete", senders=senders)
def pre_save(
senders: Union[Type["Model"], List[Type["Model"]]], **kwargs: Any
) -> Callable:
return receiver(signal="pre_save", senders=senders, **kwargs)
def pre_save(senders: Union[Type["Model"], List[Type["Model"]]],) -> Callable:
return receiver(signal="pre_save", senders=senders)
def pre_update(
senders: Union[Type["Model"], List[Type["Model"]]], **kwargs: Any
) -> Callable:
return receiver(signal="pre_update", senders=senders, **kwargs)
def pre_update(senders: Union[Type["Model"], List[Type["Model"]]]) -> Callable:
return receiver(signal="pre_update", senders=senders)
def pre_delete(
senders: Union[Type["Model"], List[Type["Model"]]], **kwargs: Any
) -> Callable:
return receiver(signal="pre_delete", senders=senders, **kwargs)
def pre_delete(senders: Union[Type["Model"], List[Type["Model"]]]) -> Callable:
return receiver(signal="pre_delete", senders=senders)