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

add most of the docstrings

Browse Source
This commit is contained in:
collerek
2021-01-04 14:43:14 +01:00
parent a914be67e2
commit eec17e2f78
8 changed files with 527 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
ormar/fields/foreign_key.py
View File

@ -122,7 +122,7 @@ def ForeignKey( # noqa CFQ002
:param kwargs: all other args to be populated by BaseField
:type kwargs: Any
:return: ormar ForeignKeyField with relation to selected model
:rtype: returns ForeignKeyField
:rtype: ForeignKeyField
"""
fk_string = to.Meta.tablename + "." + to.get_column_alias(to.Meta.pkname)
to_field = to.Meta.model_fields[to.Meta.pkname]

32
ormar/fields/many_to_many.py
View File

@ -19,6 +19,29 @@ def ManyToMany(
virtual: bool = False,
**kwargs: Any
) -> Any:
"""
Despite a name it's a function that returns constructed ManyToManyField.
This function is actually used in model declaration
(as ormar.ManyToMany(ToModel, through=ThroughModel)).
Accepts number of relation setting parameters as well as all BaseField ones.
:param to: target related ormar Model
:type to: Model class
:param through: through model for m2m relation
:type through: Model class
:param name: name of the database field - later called alias
:type name: str
:param unique: parameter passed to sqlalchemy.ForeignKey, unique flag
:type unique: bool
:param virtual: marks if relation is virtual.
It is for reversed FK and auto generated FK on through model in Many2Many relations.
:type virtual: bool
:param kwargs: all other args to be populated by BaseField
:type kwargs: Any
:return: ormar ManyToManyField with m2m relation to selected model
:rtype: ManyToManyField
"""
to_field = to.Meta.model_fields[to.Meta.pkname]
related_name = kwargs.pop("related_name", None)
nullable = kwargs.pop("nullable", True)
@ -49,8 +72,17 @@ def ManyToMany(
class ManyToManyField(ForeignKeyField, ormar.QuerySetProtocol, ormar.RelationProtocol):
"""
Actual class returned from ManyToMany function call and stored in model_fields.
"""
through: Type["Model"]
@classmethod
def default_target_field_name(cls) -> str:
"""
Returns default target model name on through model.
:return: name of the field
:rtype: str
"""
return cls.to.get_name()

42
ormar/fields/model_fields.py
View File

@ -17,6 +17,20 @@ def is_field_nullable(
server_default: Any,
pydantic_only: Optional[bool],
) -> bool:
"""
Checks if the given field should be nullable/ optional based on parameters given.
:param nullable: flag explicit setting a column as nullable
:type nullable: Optional[bool]
:param default: value or function to be called as default in python
:type default: Any
:param server_default: function to be called as default by sql server
:type server_default: Any
:param pydantic_only: flag if fields should not be included in the sql table
:type pydantic_only: Optional[bool]
:return: result of the check
:rtype: bool
"""
if nullable is None:
return (
default is not None
@ -27,10 +41,24 @@ def is_field_nullable(
def is_auto_primary_key(primary_key: bool, autoincrement: bool) -> bool:
"""
Checks if field is an autoincrement pk -> if yes it's optional.
:param primary_key: flag if field is a pk field
:type primary_key: bool
:param autoincrement: flag if field should be autoincrement
:type autoincrement: bool
:return: result of the check
:rtype: bool
"""
return primary_key and autoincrement
class ModelFieldFactory:
"""
Default field factory that construct Field classes and populated their values.
"""
_bases: Any = (BaseField,)
_type: Any = None
@ -66,10 +94,24 @@ class ModelFieldFactory:
@classmethod
def get_column_type(cls, **kwargs: Any) -> Any: # pragma no cover
"""
Return proper type of db column for given field type.
Accepts required and optional parameters that each column type accepts.
:param kwargs: key, value pairs of sqlalchemy options
:type kwargs: Any
:return: initialized column with proper options
:rtype: sqlalchemy Column
"""
return None
@classmethod
def validate(cls, **kwargs: Any) -> None: # pragma no cover
"""
Used to validate if all required parameters on a given field type are set.
:param kwargs: all params passed during construction
:type kwargs: Any
"""
pass

10
ormar/fields/sqlalchemy_uuid.py
View File

@ -10,6 +10,8 @@ class UUID(TypeDecorator): # pragma nocover
"""
Platform-independent GUID type.
Uses CHAR(36) if in a string mode, otherwise uses CHAR(32), to store UUID.
For details for different methods check documentation of parent class.
"""
impl = CHAR
@ -24,6 +26,14 @@ class UUID(TypeDecorator): # pragma nocover
return "CHAR(32)"
def _cast_to_uuid(self, value: Union[str, int, bytes]) -> uuid.UUID:
"""
Parses given value into uuid.UUID field.
:param value: value to be parsed
:type value: Union[str, int, bytes]
:return: initialized uuid
:rtype: uuid.UUID
"""
if not isinstance(value, uuid.UUID):
if isinstance(value, bytes):
ret_value = uuid.UUID(bytes=value)

241
ormar/queryset/prefetch_query.py
View File

@ -24,6 +24,18 @@ if TYPE_CHECKING: # pragma: no cover
def add_relation_field_to_fields(
fields: Union[Set[Any], Dict[Any, Any], None], related_field_name: str
) -> Union[Set[Any], Dict[Any, Any], None]:
"""
Adds related field into fields to include as otherwise it would be skipped.
Related field is added only if fields are already populated.
Empty fields implies all fields.
:param fields: Union[Set[Any], Dict[Any, Any], None]
:type fields: Dict
:param related_field_name: name of the field with relation
:type related_field_name: str
:return: updated fields dict
:rtype: Union[Set[Any], Dict[Any, Any], None]
"""
if fields and related_field_name not in fields:
if isinstance(fields, dict):
fields[related_field_name] = ...
@ -33,6 +45,18 @@ def add_relation_field_to_fields(
def sort_models(models: List["Model"], orders_by: Dict) -> List["Model"]:
"""
Since prefetch query gets all related models by ids the sorting needs to happen in
python. Since by default models are already sorted by id here we resort only if
order_by parameters was set.
:param models: list of models already fetched from db
:type models: List[tests.test_prefetch_related.Division]
:param orders_by: order by dictionary
:type orders_by: Dict[str, str]
:return: sorted list of models
:rtype: List[tests.test_prefetch_related.Division]
"""
sort_criteria = [
(key, value) for key, value in orders_by.items() if isinstance(value, str)
]
@ -54,6 +78,29 @@ def set_children_on_model( # noqa: CCR001
models: Dict,
orders_by: Dict,
) -> None:
"""
Extract ids of child models by given relation id key value.
Based on those ids the actual children model instances are fetched from
already fetched data.
If needed the child models are resorted according to passed orders_by dict.
Also relation is registered as each child is set as parent related field name value.
:param model: parent model instance
:type model: Model
:param related: name of the related field
:type related: str
:param children: dictionary of children ids/ related field value
:type children: Dict[int, set]
:param model_id: id of the model on which children should be set
:type model_id: int
:param models: dictionary of child models instances
:type models: Dict
:param orders_by: order_by dictionary
:type orders_by: Dict
"""
for key, child_models in children.items():
if key == model_id:
models_to_set = [models[child] for child in sorted(child_models)]
@ -67,6 +114,12 @@ def set_children_on_model( # noqa: CCR001
class PrefetchQuery:
"""
Query used to fetch related models in subsequent queries.
Each model is fetched only ones by the name of the relation.
That means that for each prefetch_related entry next query is issued to database.
"""
def __init__( # noqa: CFQ002
self,
model_cls: Type["Model"],
@ -92,6 +145,22 @@ class PrefetchQuery:
async def prefetch_related(
self, models: Sequence["Model"], rows: List
) -> Sequence["Model"]:
"""
Main entry point for prefetch_query.
Receives list of already initialized parent models with all children from
select_related already populated. Receives also list of row sql result rows
as it's quicker to extract ids that way instead of calling each model.
Returns list with related models already prefetched and set.
:param models: list of already instantiated models from main query
:type models: List[Model]
:param rows: row sql result of the main query before the prefetch
:type rows: List[sqlalchemy.engine.result.RowProxy]
:return: list of models with children prefetched
:rtype: List[Model]
"""
self.models = extract_models_to_dict_of_lists(
model_type=self.model, models=models, select_dict=self.select_dict
)
@ -101,6 +170,17 @@ class PrefetchQuery:
def _extract_ids_from_raw_data(
self, parent_model: Type["Model"], column_name: str
) -> Set:
"""
Iterates over raw rows and extract id values of relation columns by using
prefixed column name.
:param parent_model: ormar model class
:type parent_model: Type[Model]
:param column_name: name of the relation column which is a key column
:type column_name: str
:return: set of ids of related model that should be extracted
:rtype: set
"""
list_of_ids = set()
current_data = self.already_extracted.get(parent_model.get_name(), {})
table_prefix = current_data.get("prefix", "")
@ -113,6 +193,17 @@ class PrefetchQuery:
def _extract_ids_from_preloaded_models(
self, parent_model: Type["Model"], column_name: str
) -> Set:
"""
Extracts relation ids from already populated models if they were included
in the original query before.
:param parent_model: model from which related ids should be extracted
:type parent_model: Type["Model"]
:param column_name: name of the relation column which is a key column
:type column_name: str
:return: set of ids of related model that should be extracted
:rtype: set
"""
list_of_ids = set()
for model in self.models.get(parent_model.get_name(), []):
child = getattr(model, column_name)
@ -125,7 +216,19 @@ class PrefetchQuery:
def _extract_required_ids(
self, parent_model: Type["Model"], reverse: bool, related: str,
) -> Set:
"""
Delegates extraction of the fields to either get ids from raw sql response
or from already populated models.
:param parent_model: model from which related ids should be extracted
:type parent_model: Type["Model"]
:param reverse: flag if the relation is reverse
:type reverse: bool
:param related: name of the field with relation
:type related: str
:return: set of ids of related model that should be extracted
:rtype: set
"""
use_raw = parent_model.get_name() not in self.models
column_name = parent_model.get_column_name_for_id_extraction(
@ -151,6 +254,23 @@ class PrefetchQuery:
reverse: bool,
related: str,
) -> List:
"""
Populates where clause with condition to return only models within the
set of extracted ids.
If there are no ids for relation the empty list is returned.
:param parent_model: model from which related ids should be extracted
:type parent_model: Type["Model"]
:param target_model: model to which relation leads to
:type target_model: Type["Model"]
:param reverse: flag if the relation is reverse
:type reverse: bool
:param related: name of the field with relation
:type related: str
:return:
:rtype: List[sqlalchemy.sql.elements.TextClause]
"""
ids = self._extract_required_ids(
parent_model=parent_model, reverse=reverse, related=related
)
@ -175,7 +295,19 @@ class PrefetchQuery:
def _populate_nested_related(
self, model: "Model", prefetch_dict: Dict, orders_by: Dict,
) -> "Model":
"""
Populates all related models children of parent model that are
included in prefetch query.
:param model: ormar model instance
:type model: Model
:param prefetch_dict: dictionary of models to prefetch
:type prefetch_dict: Dict
:param orders_by: dictionary of order bys
:type orders_by: Dict
:return: model with children populated
:rtype: Model
"""
related_to_extract = model.get_filtered_names_to_extract(
prefetch_dict=prefetch_dict
)
@ -206,6 +338,24 @@ class PrefetchQuery:
async def _prefetch_related_models(
self, models: Sequence["Model"], rows: List
) -> Sequence["Model"]:
"""
Main method of the query.
Translates select nad prefetch list into dictionaries to avoid querying the
same related models multiple times.
Keeps the list of already extracted models.
Extracts the related models from the database and later populate all children
on each of the parent models from list.
:param models: list of parent models from main query
:type models: List[Model]
:param rows: raw response from sql query
:type rows: List[sqlalchemy.engine.result.RowProxy]
:return: list of models with prefetch children populated
:rtype: List[Model]
"""
self.already_extracted = {self.model.get_name(): {"raw": rows}}
select_dict = translate_list_to_dict(self._select_related)
prefetch_dict = translate_list_to_dict(self._prefetch_related)
@ -242,7 +392,32 @@ class PrefetchQuery:
exclude_fields: Union[Set[Any], Dict[Any, Any], None],
orders_by: Dict,
) -> None:
"""
Constructs queries with required ids and extracts data with fields that should
be included/excluded.
Runs the queries against the database and populated dictionaries with ids and
with actual extracted children models.
Calls itself recurrently to extract deeper nested relations of related model.
:param related: name of the relation
:type related: str
:param target_model: model to which relation leads to
:type target_model: Type[Model]
:param prefetch_dict: prefetch related list converted into dictionary
:type prefetch_dict: Dict
:param select_dict: select related list converted into dictionary
:type select_dict: Dict
:param fields: fields to include
:type fields: Union[Set[Any], Dict[Any, Any], None]
:param exclude_fields: fields to exclude
:type exclude_fields: Union[Set[Any], Dict[Any, Any], None]
:param orders_by: dictionary of order bys clauses
:type orders_by: Dict
:return: None
:rtype: None
"""
fields = target_model.get_included(fields, related)
exclude_fields = target_model.get_excluded(exclude_fields, related)
target_field = target_model.Meta.model_fields[related]
@ -320,6 +495,24 @@ class PrefetchQuery:
exclude_fields: Union[Set[Any], Dict[Any, Any], None],
filter_clauses: List,
) -> Tuple[str, List]:
"""
Actually runs the queries against the database and populates the raw response
for given related model.
Returns table prefix as it's later needed to eventually initialize the children
models.
:param target_field: ormar field with relation definition
:type target_field: Type["BaseField"]
:param fields: fields to include
:type fields: Union[Set[Any], Dict[Any, Any], None]
:param exclude_fields: fields to exclude
:type exclude_fields: Union[Set[Any], Dict[Any, Any], None]
:param filter_clauses: list of clauses, actually one clause with ids of relation
:type filter_clauses: List[sqlalchemy.sql.elements.TextClause]
:return: table prefix and raw rows from sql response
:rtype: Tuple[str, List]
"""
target_model = target_field.to
target_name = target_model.get_name()
select_related = []
@ -353,6 +546,17 @@ class PrefetchQuery:
@staticmethod
def _get_select_related_if_apply(related: str, select_dict: Dict) -> Dict:
"""
Extract nested part of select_related dictionary to extract models nested
deeper on related model and already loaded in select related query.
:param related: name of the relation
:type related: str
:param select_dict: dictionary of select related models in main query
:type select_dict: Dict
:return: dictionary with nested part of select related
:rtype: Dict
"""
return (
select_dict.get(related, {})
if (select_dict and select_dict is not Ellipsis and related in select_dict)
@ -362,6 +566,16 @@ class PrefetchQuery:
def _update_already_loaded_rows( # noqa: CFQ002
self, target_field: Type["BaseField"], prefetch_dict: Dict, orders_by: Dict,
) -> None:
"""
Updates models that are already loaded, usually children of children.
:param target_field: ormar field with relation definition
:type target_field: Type["BaseField"]
:param prefetch_dict: dictionaries of related models to prefetch
:type prefetch_dict: Dict
:param orders_by: dictionary of order by clauses by model
:type orders_by: Dict
"""
target_model = target_field.to
for instance in self.models.get(target_model.get_name(), []):
self._populate_nested_related(
@ -379,6 +593,33 @@ class PrefetchQuery:
prefetch_dict: Dict,
orders_by: Dict,
) -> None:
"""
Instantiates children models extracted from given relation.
Populates them with their own nested children if they are included in prefetch
query.
Sets the initialized models and ids of them under corresponding keys in
already_extracted dictionary. Later those instances will be fetched by ids
and set on the parent model after sorting if needed.
:param rows: raw sql response from the prefetch query
:type rows: List[sqlalchemy.engine.result.RowProxy]
:param target_field: field with relation definition from parent model
:type target_field: Type["BaseField"]
:param parent_model: model with relation definition
:type parent_model: Type[Model]
:param table_prefix: prefix of the target table from current relation
:type table_prefix: str
:param fields: fields to include
:type fields: Union[Set[Any], Dict[Any, Any], None]
:param exclude_fields: fields to exclude
:type exclude_fields: Union[Set[Any], Dict[Any, Any], None]
:param prefetch_dict: dictionaries of related models to prefetch
:type prefetch_dict: Dict
:param orders_by: dictionary of order by clauses by model
:type orders_by: Dict
"""
target_model = target_field.to
for row in rows:
field_name = parent_model.get_related_field_name(target_field=target_field)

53
ormar/queryset/query.py
View File

@ -49,19 +49,41 @@ class Query:
self.limit_raw_sql = limit_raw_sql
def _init_sorted_orders(self) -> None:
"""
Initialize empty order_by dict to be populated later during the query call
"""
if self.order_columns:
for clause in self.order_columns:
self.sorted_orders[clause] = None
@property
def prefixed_pk_name(self) -> str:
"""
Shortcut for extracting prefixed with alias primary key column name from main
model
:return: alias of pk column prefix with table name.
:rtype: str
"""
pkname_alias = self.model_cls.get_column_alias(self.model_cls.Meta.pkname)
return f"{self.table.name}.{pkname_alias}"
def alias(self, name: str) -> str:
"""
Shortcut to extracting column alias from given master model.
:param name: name of column
:type name: str
:return: alias of given column name
:rtype: str
"""
return self.model_cls.get_column_alias(name)
def apply_order_bys_for_primary_model(self) -> None: # noqa: CCR001
"""
Applies order_by queries on main model when it's used as a subquery.
That way the subquery with limit and offset only on main model has proper
sorting applied and correct models are fetched.
"""
if self.order_columns:
for clause in self.order_columns:
if "__" not in clause:
@ -91,6 +113,18 @@ class Query:
)
def build_select_expression(self) -> Tuple[sqlalchemy.sql.select, List[str]]:
"""
Main entry point from outside (after proper initialization).
Extracts columns list to fetch,
construct all required joins for select related,
then applies all conditional and sort clauses.
Returns ready to run query with all joins and clauses.
:return: ready to run query with all joins and clauses.
:rtype: sqlalchemy.sql.selectable.Select
"""
self_related_fields = self.model_cls.own_table_columns(
model=self.model_cls,
fields=self.fields,
@ -184,6 +218,21 @@ class Query:
def _apply_expression_modifiers(
self, expr: sqlalchemy.sql.select
) -> sqlalchemy.sql.select:
"""
Receives the select query (might be join) and applies:
* Filter clauses
* Exclude filter clauses
* Limit clauses
* Offset clauses
* Order by clauses
Returns complete ready to run query.
:param expr: select expression before clauses
:type expr: sqlalchemy.sql.selectable.Select
:return: expresion with all present clauses applied
:rtype: sqlalchemy.sql.selectable.Select
"""
expr = FilterQuery(filter_clauses=self.filter_clauses).apply(expr)
expr = FilterQuery(filter_clauses=self.exclude_clauses, exclude=True).apply(
expr
@ -195,6 +244,10 @@ class Query:
return expr
def _reset_query_parameters(self) -> None:
"""
Although it should be created each time before the call we reset the key params
anyway.
"""
self.select_from = []
self.columns = []
self.used_aliases = []

99
ormar/queryset/utils.py
View File

@ -18,6 +18,22 @@ if TYPE_CHECKING: # pragma no cover
def check_node_not_dict_or_not_last_node(
part: str, parts: List, current_level: Any
) -> bool:
"""
Checks if given name is not present in the current level of the structure.
Checks if given name is not the last name in the split list of parts.
Checks if the given name in current level is not a dictionary.
All those checks verify if there is a need for deeper traversal.
:param part:
:type part: str
:param parts:
:type parts: List[str]
:param current_level: current level of the traversed structure
:type current_level: Any
:return: result of the check
:rtype: bool
"""
return (part not in current_level and part != parts[-1]) or (
part in current_level and not isinstance(current_level[part], dict)
)
@ -26,6 +42,21 @@ def check_node_not_dict_or_not_last_node(
def translate_list_to_dict( # noqa: CCR001
list_to_trans: Union[List, Set], is_order: bool = False
) -> Dict:
"""
Splits the list of strings by '__' and converts them to dictionary with nested
models grouped by parent model. That way each model appears only once in the whole
dictionary and children are grouped under parent name.
Default required key ise Ellipsis like in pydantic.
:param list_to_trans: input list
:type list_to_trans: set
:param is_order: flag if change affects order_by clauses are they require special
default value with sort order.
:type is_order: bool
:return: converted to dictionary input list
:rtype: Dict
"""
new_dict: Dict = dict()
for path in list_to_trans:
current_level = new_dict
@ -50,6 +81,15 @@ def translate_list_to_dict( # noqa: CCR001
def convert_set_to_required_dict(set_to_convert: set) -> Dict:
"""
Converts set to dictionary of required keys.
Required key is Ellipsis.
:param set_to_convert: set to convert to dict
:type set_to_convert: set
:return: set converted to dict of ellipsis
:rtype: Dict[str, ellipsis]
"""
new_dict = dict()
for key in set_to_convert:
new_dict[key] = Ellipsis
@ -57,6 +97,19 @@ def convert_set_to_required_dict(set_to_convert: set) -> Dict:
def update(current_dict: Any, updating_dict: Any) -> Dict: # noqa: CCR001
"""
Update one dict with another but with regard for nested keys.
That way nested sets are unionised, dicts updated and
only other values are overwritten.
:param current_dict: dict to update
:type current_dict: Dict[str, ellipsis]
:param updating_dict: dict with values to update
:type updating_dict: Dict
:return: combination of both dicts
:rtype: Dict
"""
if current_dict is Ellipsis:
current_dict = dict()
for key, value in updating_dict.items():
@ -73,6 +126,17 @@ def update(current_dict: Any, updating_dict: Any) -> Dict: # noqa: CCR001
def update_dict_from_list(curr_dict: Dict, list_to_update: Union[List, Set]) -> Dict:
"""
Converts the list into dictionary and later performs special update, where
nested keys that are sets or dicts are combined and not overwritten.
:param curr_dict: dict to update
:type curr_dict: Dict
:param list_to_update: list with values to update the dict
:type list_to_update: List[str]
:return: updated dict
:rtype: Dict
"""
updated_dict = copy.copy(curr_dict)
dict_to_update = translate_list_to_dict(list_to_update)
update(updated_dict, dict_to_update)
@ -82,6 +146,25 @@ def update_dict_from_list(curr_dict: Dict, list_to_update: Union[List, Set]) ->
def extract_nested_models( # noqa: CCR001
model: "Model", model_type: Type["Model"], select_dict: Dict, extracted: Dict
) -> None:
"""
Iterates over model relations and extracts all nested models from select_dict and
puts them in corresponding list under relation name in extracted dict.keys
Basically flattens all relation to dictionary of all related models, that can be
used on several models and extract all of their children into dictionary of lists
witch children models.
Goes also into nested relations if needed (specified in select_dict).
:param model: parent Model
:type model: Model
:param model_type: parent model class
:type model_type: Type[Model]
:param select_dict: dictionary of related models from select_related
:type select_dict: Dict
:param extracted: dictionary with already extracted models
:type extracted: Dict
"""
follow = [rel for rel in model_type.extract_related_names() if rel in select_dict]
for related in follow:
child = getattr(model, related)
@ -108,6 +191,22 @@ def extract_models_to_dict_of_lists(
select_dict: Dict,
extracted: Dict = None,
) -> Dict:
"""
Receives a list of models and extracts all of the children and their children
into dictionary of lists with children models, flattening the structure to one dict
with all children models under their relation keys.
:param model_type: parent model class
:type model_type: Type[Model]
:param models: list of models from which related models should be extracted.
:type models: List[Model]
:param select_dict: dictionary of related models from select_related
:type select_dict: Dict
:param extracted: dictionary with already extracted models
:type extracted: Dict
:return: dictionary of lists f related models
:rtype: Dict
"""
if not extracted:
extracted = dict()
for model in models:

49
ormar/signals/signal.py
View File

@ -9,6 +9,14 @@ if TYPE_CHECKING: # pragma: no cover
def callable_accepts_kwargs(func: Callable) -> bool:
"""
Checks if function accepts **kwargs.
:param func: function which signature needs to be checked
:type func: function
:return:
:rtype: bool
"""
return any(
p
for p in inspect.signature(func).parameters.values()
@ -17,16 +25,37 @@ def callable_accepts_kwargs(func: Callable) -> bool:
def make_id(target: Any) -> Union[int, Tuple[int, int]]:
"""
Creates id of a function or method to be used as key to store signal
:param target: target which id we want
:type target: Any
:return: id of the target
:rtype: int
"""
if hasattr(target, "__func__"):
return id(target.__self__), id(target.__func__)
return id(target)
class Signal:
"""
Signal that notifies all receiver functions.
In ormar used by models to send pre_save, post_save etc. signals.
"""
def __init__(self) -> None:
self._receivers: List[Tuple[Union[int, Tuple[int, int]], Callable]] = []
def connect(self, receiver: Callable) -> None:
"""
Connects given receiver function to the signal.
:raises: SignalDefinitionError if receiver is not callable
or not accept **kwargs
:param receiver: receiver function
:type receiver: Callable
"""
if not callable(receiver):
raise SignalDefinitionError("Signal receivers must be callable.")
if not callable_accepts_kwargs(receiver):
@ -38,6 +67,14 @@ class Signal:
self._receivers.append((new_receiver_key, receiver))
def disconnect(self, receiver: Callable) -> bool:
"""
Removes the receiver function from the signal.
:param receiver: receiver function
:type receiver: Callable
:return: flag if receiver was removed
:rtype: bool
"""
removed = False
new_receiver_key = make_id(receiver)
for ind, rec in enumerate(self._receivers):
@ -49,6 +86,13 @@ class Signal:
return removed
async def send(self, sender: Type["Model"], **kwargs: Any) -> None:
"""
Notifies all receiver functions with given kwargs
:param sender: model that sends the signal
:type sender: Type["Model"]
:param kwargs: arguments passed to receivers
:type kwargs: Any
"""
receivers = []
for receiver in self._receivers:
_, receiver_func = receiver
@ -57,6 +101,11 @@ class Signal:
class SignalEmitter:
"""
Emitter that registers the signals in internal dictionary.
If signal with given name does not exist it's auto added on access.
"""
if TYPE_CHECKING: # pragma: no cover
signals: Dict[str, Signal]