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

finish docstrings in relations package

Browse Source
This commit is contained in:
collerek
2021-01-03 16:46:46 +01:00
parent 7a8d11b1c7
commit a32a3b9d59
7 changed files with 623 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
docs/queries.md
View File

@ -47,7 +47,7 @@ await malibu.save()
Get's the first row from the db meeting the criteria set by kwargs.
If no criteria set it will return the first row in db.
If no criteria set it will return the last row in db sorted by pk.
Passing a criteria is actually calling filter(**kwargs) method described below.
@ -86,6 +86,13 @@ assert album == album2
!!!note
Note that if you want to create a new object you either have to pass pk column value or pk column has to be set as autoincrement
### first
`first(): -> Model`
Gets the first row from the db ordered by primary key column ascending.
### update
`update(each: bool = False, **kwargs) -> int`
@ -447,9 +454,12 @@ any attribute it will be updated on all parents as they share the same child obj
### limit
`limit(limit_count: int) -> QuerySet`
`limit(limit_count: int, limit_raw_sql: bool = None) -> QuerySet`
You can limit the results to desired number of rows.
You can limit the results to desired number of parent models.
To limit the actual number of database query rows instead of number of main models
use the `limit_raw_sql` parameter flag, and set it to `True`.
```python
tracks = await Track.objects.limit(1).all()
@ -465,9 +475,12 @@ tracks = await Track.objects.limit(1).all()
### offset
`offset(offset: int) -> QuerySet`
`offset(offset: int, limit_raw_sql: bool = None) -> QuerySet`
You can also offset the results by desired number of rows.
You can also offset the results by desired number of main models.
To offset the actual number of database query rows instead of number of main models
use the `limit_raw_sql` parameter flag, and set it to `True`.
```python
tracks = await Track.objects.offset(1).limit(1).all()

4
ormar/relations/__init__.py
View File

@ -1,3 +1,7 @@
"""
Package handles relations on models, returning related models on calls and exposing
QuerySetProxy for m2m and reverse relations.
"""
from ormar.relations.alias_manager import AliasManager
from ormar.relations.relation import Relation, RelationType
from ormar.relations.relation_manager import RelationsManager

367
ormar/relations/querysetproxy.py
View File

@ -23,6 +23,11 @@ if TYPE_CHECKING: # pragma no cover
class QuerysetProxy(ormar.QuerySetProtocol):
"""
Exposes QuerySet methods on relations, but also handles creating and removing
of through Models for m2m relations.
"""
if TYPE_CHECKING: # pragma no cover
relation: "Relation"
@ -42,21 +47,43 @@ class QuerysetProxy(ormar.QuerySetProtocol):
@property
def queryset(self) -> "QuerySet":
"""
Returns queryset if it's set, AttributeError otherwise.
:return: QuerySet
:rtype: QuerySet
"""
if not self._queryset:
raise AttributeError
return self._queryset
@queryset.setter
def queryset(self, value: "QuerySet") -> None:
"""
Set's the queryset. Initialized in RelationProxy.
:param value: QuerySet
:type value: QuerySet
"""
self._queryset = value
def _assign_child_to_parent(self, child: Optional["T"]) -> None:
"""
Registers child in parents RelationManager.
:param child: child to register on parent side.
:type child: Model
"""
if child:
owner = self._owner
rel_name = self.relation.field_name
setattr(owner, rel_name, child)
def _register_related(self, child: Union["T", Sequence[Optional["T"]]]) -> None:
"""
Registers child/ children in parents RelationManager.
:param child: child or list of children models to register.
:type child: Union[Model,List[Model]]
"""
if isinstance(child, list):
for subchild in child:
self._assign_child_to_parent(subchild)
@ -65,11 +92,20 @@ class QuerysetProxy(ormar.QuerySetProtocol):
self._assign_child_to_parent(child)
def _clean_items_on_load(self) -> None:
"""
Cleans the current list of the related models.
"""
if isinstance(self.relation.related_models, MutableSequence):
for item in self.relation.related_models[:]:
self.relation.remove(item)
async def create_through_instance(self, child: "T") -> None:
"""
Crete a through model instance in the database for m2m relations.
:param child: child model instance
:type child: Model
"""
queryset = ormar.QuerySet(model_cls=self.relation.through)
owner_column = self._owner.get_name()
child_column = child.get_name()
@ -77,6 +113,12 @@ class QuerysetProxy(ormar.QuerySetProtocol):
await queryset.create(**kwargs)
async def delete_through_instance(self, child: "T") -> None:
"""
Removes through model instance from the database for m2m relations.
:param child: child model instance
:type child: Model
"""
queryset = ormar.QuerySet(model_cls=self.relation.through)
owner_column = self._owner.get_name()
child_column = child.get_name()
@ -85,12 +127,45 @@ class QuerysetProxy(ormar.QuerySetProtocol):
await link_instance.delete()
async def exists(self) -> bool:
"""
Returns a bool value to confirm if there are rows matching the given criteria
(applied with `filter` and `exclude` if set).
Actual call delegated to QuerySet.
:return: result of the check
:rtype: bool
"""
return await self.queryset.exists()
async def count(self) -> int:
"""
Returns number of rows matching the given criteria
(applied with `filter` and `exclude` if set before).
Actual call delegated to QuerySet.
:return: number of rows
:rtype: int
"""
return await self.queryset.count()
async def clear(self, keep_reversed: bool = True) -> int:
"""
Removes all related models from given relation.
Removes all through models for m2m relation.
For reverse FK relations keep_reversed flag marks if the reversed models
should be kept or deleted from the database too (False means that models
will be deleted, and not only removed from relation).
:param keep_reversed: flag if reverse models in reverse FK should be deleted
or not, keep_reversed=False deletes them from database.
:type keep_reversed: bool
:return: number of deleted models
:rtype: int
"""
if self.type_ == ormar.RelationType.MULTIPLE:
queryset = ormar.QuerySet(model_cls=self.relation.through)
owner_column = self._owner.get_name()
@ -107,24 +182,85 @@ class QuerysetProxy(ormar.QuerySetProtocol):
return await queryset.delete(**kwargs) # type: ignore
async def first(self, **kwargs: Any) -> "Model":
"""
Gets the first row from the db ordered by primary key column ascending.
Actual call delegated to QuerySet.
List of related models is cleared before the call.
:param kwargs:
:type kwargs:
:return:
:rtype: _asyncio.Future
"""
first = await self.queryset.first(**kwargs)
self._clean_items_on_load()
self._register_related(first)
return first
async def get(self, **kwargs: Any) -> "Model":
"""
Get's the first row from the db meeting the criteria set by kwargs.
If no criteria set it will return the last row in db sorted by pk.
Passing a criteria is actually calling filter(**kwargs) method described below.
Actual call delegated to QuerySet.
List of related models is cleared before the call.
:raises: NoMatch if no rows are returned
:raises: MultipleMatches if more than 1 row is returned.
:param kwargs: fields names and proper value types
:type kwargs: Any
:return: returned model
:rtype: Model
"""
get = await self.queryset.get(**kwargs)
self._clean_items_on_load()
self._register_related(get)
return get
async def all(self, **kwargs: Any) -> Sequence[Optional["Model"]]: # noqa: A003
"""
Returns all rows from a database for given model for set filter options.
Passing kwargs is a shortcut and equals to calling `filter(**kwrags).all()`.
If there are no rows meeting the criteria an empty list is returned.
Actual call delegated to QuerySet.
List of related models is cleared before the call.
:param kwargs: fields names and proper value types
:type kwargs: Any
:return: list of returned models
:rtype: List[Model]
"""
all_items = await self.queryset.all(**kwargs)
self._clean_items_on_load()
self._register_related(all_items)
return all_items
async def create(self, **kwargs: Any) -> "Model":
"""
Creates the model instance, saves it in a database and returns the updates model
(with pk populated if not passed and autoincrement is set).
The allowed kwargs are `Model` fields names and proper value types.
For m2m relation the through model is created automatically.
Actual call delegated to QuerySet.
:param kwargs: fields names and proper value types
:type kwargs: Any
:return: created model
:rtype: Model
"""
if self.type_ == ormar.RelationType.REVERSE:
kwargs[self.related_field.name] = self._owner
created = await self.queryset.create(**kwargs)
@ -134,12 +270,34 @@ class QuerysetProxy(ormar.QuerySetProtocol):
return created
async def get_or_create(self, **kwargs: Any) -> "Model":
"""
Combination of create and get methods.
Tries to get a row meeting the criteria fro kwargs
and if `NoMatch` exception is raised
it creates a new one with given kwargs.
:param kwargs: fields names and proper value types
:type kwargs: Any
:return: returned or created Model
:rtype: Model
"""
try:
return await self.get(**kwargs)
except ormar.NoMatch:
return await self.create(**kwargs)
async def update_or_create(self, **kwargs: Any) -> "Model":
"""
Updates the model, or in case there is no match in database creates a new one.
Actual call delegated to QuerySet.
:param kwargs: fields names and proper value types
:type kwargs: Any
:return: updated or created model
:rtype: Model
"""
pk_name = self.queryset.model_meta.pkname
if "pk" in kwargs:
kwargs[pk_name] = kwargs.pop("pk")
@ -149,37 +307,246 @@ class QuerysetProxy(ormar.QuerySetProtocol):
return await model.update(**kwargs)
def filter(self, **kwargs: Any) -> "QuerysetProxy": # noqa: A003, A001
"""
Allows you to filter by any `Model` attribute/field
as well as to fetch instances, with a filter across an FK relationship.
You can use special filter suffix to change the filter operands:
* exact - like `album__name__exact='Malibu'` (exact match)
* iexact - like `album__name__iexact='malibu'` (exact match case insensitive)
* contains - like `album__name__contains='Mal'` (sql like)
* icontains - like `album__name__icontains='mal'` (sql like case insensitive)
* in - like `album__name__in=['Malibu', 'Barclay']` (sql in)
* gt - like `position__gt=3` (sql >)
* gte - like `position__gte=3` (sql >=)
* lt - like `position__lt=3` (sql <)
* lte - like `position__lte=3` (sql <=)
* startswith - like `album__name__startswith='Mal'` (exact start match)
* istartswith - like `album__name__istartswith='mal'` (case insensitive)
* endswith - like `album__name__endswith='ibu'` (exact end match)
* iendswith - like `album__name__iendswith='IBU'` (case insensitive)
Actual call delegated to QuerySet.
:param kwargs: fields names and proper value types
:type kwargs: Any
:return: filtered QuerysetProxy
:rtype: QuerysetProxy
"""
queryset = self.queryset.filter(**kwargs)
return self.__class__(relation=self.relation, type_=self.type_, qryset=queryset)
def exclude(self, **kwargs: Any) -> "QuerysetProxy": # noqa: A003, A001
"""
Works exactly the same as filter and all modifiers (suffixes) are the same,
but returns a *not* condition.
So if you use `filter(name='John')` which is `where name = 'John'` in SQL,
the `exclude(name='John')` equals to `where name <> 'John'`
Note that all conditions are joined so if you pass multiple values it
becomes a union of conditions.
`exclude(name='John', age>=35)` will become
`where not (name='John' and age>=35)`
Actual call delegated to QuerySet.
:param kwargs: fields names and proper value types
:type kwargs: Any
:return: filtered QuerysetProxy
:rtype: QuerysetProxy
"""
queryset = self.queryset.exclude(**kwargs)
return self.__class__(relation=self.relation, type_=self.type_, qryset=queryset)
def select_related(self, related: Union[List, str]) -> "QuerysetProxy":
"""
Allows to prefetch related models during the same query.
**With `select_related` always only one query is run against the database**,
meaning that one (sometimes complicated) join is generated and later nested
models are processed in python.
To fetch related model use `ForeignKey` names.
To chain related `Models` relation use double underscores between names.
Actual call delegated to QuerySet.
:param related: list of relation field names, can be linked by '__' to nest
:type related: str
:return: QuerysetProxy
:rtype: QuerysetProxy
"""
queryset = self.queryset.select_related(related)
return self.__class__(relation=self.relation, type_=self.type_, qryset=queryset)
def prefetch_related(self, related: Union[List, str]) -> "QuerysetProxy":
"""
Allows to prefetch related models during query - but opposite to
`select_related` each subsequent model is fetched in a separate database query.
**With `prefetch_related` always one query per Model is run against the
database**, meaning that you will have multiple queries executed one
after another.
To fetch related model use `ForeignKey` names.
To chain related `Models` relation use double underscores between names.
Actual call delegated to QuerySet.
:param related: list of relation field names, can be linked by '__' to nest
:type related: str
:return: QuerysetProxy
:rtype: QuerysetProxy
"""
queryset = self.queryset.prefetch_related(related)
return self.__class__(relation=self.relation, type_=self.type_, qryset=queryset)
def limit(self, limit_count: int) -> "QuerysetProxy":
"""
You can limit the results to desired number of parent models.
Actual call delegated to QuerySet.
:param limit_count: number of models to limit
:type limit_count: int
:return: QuerysetProxy
:rtype: QuerysetProxy
"""
queryset = self.queryset.limit(limit_count)
return self.__class__(relation=self.relation, type_=self.type_, qryset=queryset)
def offset(self, offset: int) -> "QuerysetProxy":
"""
You can also offset the results by desired number of main models.
Actual call delegated to QuerySet.
:param offset: numbers of models to offset
:type offset: int
:return: QuerysetProxy
:rtype: QuerysetProxy
"""
queryset = self.queryset.offset(offset)
return self.__class__(relation=self.relation, type_=self.type_, qryset=queryset)
def fields(self, columns: Union[List, str, Set, Dict]) -> "QuerysetProxy":
"""
With `fields()` you can select subset of model columns to limit the data load.
Note that `fields()` and `exclude_fields()` works both for main models
(on normal queries like `get`, `all` etc.)
as well as `select_related` and `prefetch_related`
models (with nested notation).
You can select specified fields by passing a `str, List[str], Set[str] or
dict` with nested definition.
To include related models use notation
`{related_name}__{column}[__{optional_next} etc.]`.
`fields()` can be called several times, building up the columns to select.
If you include related models into `select_related()` call but you won't specify
columns for those models in fields - implies a list of all fields for
those nested models.
Mandatory fields cannot be excluded as it will raise `ValidationError`,
to exclude a field it has to be nullable.
Pk column cannot be excluded - it's always auto added even if
not explicitly included.
You can also pass fields to include as dictionary or set.
To mark a field as included in a dictionary use it's name as key
and ellipsis as value.
To traverse nested models use nested dictionaries.
To include fields at last level instead of nested dictionary a set can be used.
To include whole nested model specify model related field name and ellipsis.
Actual call delegated to QuerySet.
:param columns: columns to include
:type columns: Union[List, str, Set, Dict]
:return: QuerysetProxy
:rtype: QuerysetProxy
"""
queryset = self.queryset.fields(columns)
return self.__class__(relation=self.relation, type_=self.type_, qryset=queryset)
def exclude_fields(self, columns: Union[List, str, Set, Dict]) -> "QuerysetProxy":
"""
With `exclude_fields()` you can select subset of model columns that will
be excluded to limit the data load.
It's the opposite of `fields()` method so check documentation above
to see what options are available.
Especially check above how you can pass also nested dictionaries
and sets as a mask to exclude fields from whole hierarchy.
Note that `fields()` and `exclude_fields()` works both for main models
(on normal queries like `get`, `all` etc.)
as well as `select_related` and `prefetch_related` models
(with nested notation).
Mandatory fields cannot be excluded as it will raise `ValidationError`,
to exclude a field it has to be nullable.
Pk column cannot be excluded - it's always auto added even
if explicitly excluded.
Actual call delegated to QuerySet.
:param columns: columns to exclude
:type columns: Union[List, str, Set, Dict]
:return: QuerysetProxy
:rtype: QuerysetProxy
"""
queryset = self.queryset.exclude_fields(columns=columns)
return self.__class__(relation=self.relation, type_=self.type_, qryset=queryset)
def order_by(self, columns: Union[List, str]) -> "QuerysetProxy":
"""
With `order_by()` you can order the results from database based on your
choice of fields.
You can provide a string with field name or list of strings with fields names.
Ordering in sql will be applied in order of names you provide in order_by.
By default if you do not provide ordering `ormar` explicitly orders by
all primary keys
If you are sorting by nested models that causes that the result rows are
unsorted by the main model `ormar` will combine those children rows into
one main model.
The main model will never duplicate in the result
To order by main model field just provide a field name
To sort on nested models separate field names with dunder '__'.
You can sort this way across all relation types -> `ForeignKey`,
reverse virtual FK and `ManyToMany` fields.
To sort in descending order provide a hyphen in front of the field name
Actual call delegated to QuerySet.
:param columns: columns by which models should be sorted
:type columns: Union[List, str]
:return: QuerysetProxy
:rtype: QuerysetProxy
"""
queryset = self.queryset.order_by(columns)
return self.__class__(relation=self.relation, type_=self.type_, qryset=queryset)

59
ormar/relations/relation.py
View File

@ -15,12 +15,23 @@ if TYPE_CHECKING: # pragma no cover
class RelationType(Enum):
"""
Different types of relations supported by ormar.
ForeignKey = PRIMARY
reverse ForeignKey = REVERSE
ManyToMany = MULTIPLE
"""
PRIMARY = 1
REVERSE = 2
MULTIPLE = 3
class Relation:
"""
Keeps related Models and handles adding/removing of the children.
"""
def __init__(
self,
manager: "RelationsManager",
@ -29,6 +40,23 @@ class Relation:
to: Type["T"],
through: Type["T"] = None,
) -> None:
"""
Initialize the Relation and keep the related models either as instances of
passed Model, or as a RelationProxy which is basically a list of models with
some special behavior, as it exposes QuerySetProxy and allows querying the
related models already pre filtered by parent model.
:param manager: reference to relation manager
:type manager: RelationsManager
:param type_: type of the relation
:type type_: RelationType
:param field_name: name of the relation field
:type field_name: str
:param to: model to which relation leads to
:type to: Type[Model]
:param through: model through which relation goes for m2m relations
:type through: Type[Model]
"""
self.manager = manager
self._owner: "Model" = manager.owner
self._type: RelationType = type_
@ -43,6 +71,9 @@ class Relation:
)
def _clean_related(self) -> None:
"""
Removes dead weakrefs from RelationProxy.
"""
cleaned_data = [
x
for i, x in enumerate(self.related_models) # type: ignore
@ -61,6 +92,14 @@ class Relation:
def _find_existing(
self, child: Union["NewBaseModel", Type["NewBaseModel"]]
) -> Optional[int]:
"""
Find child model in RelationProxy if exists.
:param child: child model to find
:type child: Model
:return: index of child in RelationProxy
:rtype: Optional[ind]
"""
if not isinstance(self.related_models, RelationProxy): # pragma nocover
raise ValueError("Cannot find existing models in parent relation type")
if self._to_remove:
@ -74,6 +113,13 @@ class Relation:
return None
def add(self, child: "T") -> None:
"""
Adds child Model to relation, either sets child as related model or adds
it to the list in RelationProxy depending on relation type.
:param child: model to add to relation
:type child: Model
"""
relation_name = self.field_name
if self._type == RelationType.PRIMARY:
self.related_models = child
@ -89,6 +135,13 @@ class Relation:
self._owner.__dict__[relation_name] = rel
def remove(self, child: Union["NewBaseModel", Type["NewBaseModel"]]) -> None:
"""
Removes child Model from relation, either sets None as related model or removes
it from the list in RelationProxy depending on relation type.
:param child: model to remove from relation
:type child: Model
"""
relation_name = self.field_name
if self._type == RelationType.PRIMARY:
if self.related_models == child:
@ -101,6 +154,12 @@ class Relation:
del self._owner.__dict__[relation_name][position]
def get(self) -> Optional[Union[List["T"], "T"]]:
"""
Return the related model or models from RelationProxy.
:return: related model/models if set
:rtype: Optional[Union[List[Model], Model]]
"""
return self.related_models
def __repr__(self) -> str: # pragma no cover

85
ormar/relations/relation_manager.py
View File

@ -15,6 +15,10 @@ if TYPE_CHECKING: # pragma no cover
class RelationsManager:
"""
Manages relations on a Model, each Model has it's own instance.
"""
def __init__(
self,
related_fields: List[Type[ForeignKeyField]] = None,
@ -28,11 +32,26 @@ class RelationsManager:
self._add_relation(field)
def _get_relation_type(self, field: Type[BaseField]) -> RelationType:
"""
Returns type of the relation declared on a field.
:param field: field with relation declaration
:type field: Type[BaseField]
:return: type of the relation defined on field
:rtype: RelationType
"""
if issubclass(field, ManyToManyField):
return RelationType.MULTIPLE
return RelationType.PRIMARY if not field.virtual else RelationType.REVERSE
def _add_relation(self, field: Type[BaseField]) -> None:
"""
Registers relation in the manager.
Adds Relation instance under field.name.
:param field: field with relation declaration
:type field: Type[BaseField]
"""
self._relations[field.name] = Relation(
manager=self,
type_=self._get_relation_type(field),
@ -42,15 +61,40 @@ class RelationsManager:
)
def __contains__(self, item: str) -> bool:
"""
Checks if relation with given name is already registered.
:param item: name of attribute
:type item: str
:return: result of the check
:rtype: bool
"""
return item in self._related_names
def get(self, name: str) -> Optional[Union["T", Sequence["T"]]]:
"""
Returns the related model/models if relation is set.
Actual call is delegated to Relation instance registered under relation name.
:param name: name of the relation
:type name: str
:return: related model or list of related models if set
:rtype: Optional[Union[Model, List[Model]]
"""
relation = self._relations.get(name, None)
if relation is not None:
return relation.get()
return None # pragma nocover
def _get(self, name: str) -> Optional[Relation]:
"""
Returns the actual relation and not the related model(s).
:param name: name of the relation
:type name: str
:return: Relation instance
:rtype: ormar.relations.relation.Relation
"""
relation = self._relations.get(name, None)
if relation is not None:
return relation
@ -64,6 +108,25 @@ class RelationsManager:
virtual: bool,
relation_name: str,
) -> None:
"""
Adds relation on both sides -> meaning on both child and parent models.
One side of the relation is always weakref proxy to avoid circular refs.
Based on the side from which relation is added and relation name actual names
of parent and child relations are established. The related models are registered
on both ends.
:param parent: parent model on which relation should be registered
:type parent: Model
:param child: child model to register
:type child: Model
:param child_name: potential child name used if related name is not set
:type child_name: str
:param virtual:
:type virtual: bool
:param relation_name: name of the relation
:type relation_name: str
"""
to_field: Type[BaseField] = child.Meta.model_fields[relation_name]
# print('comming', child_name, relation_name)
(parent, child, child_name, to_name,) = get_relations_sides_and_names(
@ -83,6 +146,16 @@ class RelationsManager:
def remove(
self, name: str, child: Union["NewBaseModel", Type["NewBaseModel"]]
) -> None:
"""
Removes given child from relation with given name.
Since you can have many relations between two models you need to pass a name
of relation from which you want to remove the child.
:param name: name of the relation
:type name: str
:param child: child to remove from relation
:type child: Union[Model, Type[Model]]
"""
relation = self._get(name)
if relation:
relation.remove(child)
@ -91,6 +164,18 @@ class RelationsManager:
def remove_parent(
item: Union["NewBaseModel", Type["NewBaseModel"]], parent: "Model", name: str
) -> None:
"""
Removes given parent from relation with given name.
Since you can have many relations between two models you need to pass a name
of relation from which you want to remove the parent.
:param item: model with parent registered
:type item: Union[Model, Type[Model]]
:param parent: parent Model
:type parent: Model
:param name: name of the relation
:type name: str
"""
relation_name = (
item.Meta.model_fields[name].related_name or item.get_name() + "s"
)

72
ormar/relations/relation_proxy.py
View File

@ -11,6 +11,10 @@ if TYPE_CHECKING: # pragma no cover
class RelationProxy(list):
"""
Proxy of the Relation that is a list with special methods.
"""
def __init__(
self,
relation: "Relation",
@ -28,6 +32,13 @@ class RelationProxy(list):
@property
def related_field_name(self) -> str:
"""
On first access calculates the name of the related field, later stored in
_related_field_name property.
:return: name of the related field
:rtype: str
"""
if self._related_field_name:
return self._related_field_name
owner_field = self._owner.Meta.model_fields[self.field_name]
@ -37,26 +48,55 @@ class RelationProxy(list):
return self._related_field_name
def __getattribute__(self, item: str) -> Any:
"""
Since some QuerySetProxy methods overwrite builtin list methods we
catch calls to them and delegate it to QuerySetProxy instead.
:param item: name of attribute
:type item: str
:return: value of attribute
:rtype: Any
"""
if item in ["count", "clear"]:
self._initialize_queryset()
return getattr(self.queryset_proxy, item)
return super().__getattribute__(item)
def __getattr__(self, item: str) -> Any:
"""
Delegates calls for non existing attributes to QuerySetProxy.
:param item: name of attribute/method
:type item: str
:return: method from QuerySetProxy if exists
:rtype: method
"""
self._initialize_queryset()
return getattr(self.queryset_proxy, item)
def _initialize_queryset(self) -> None:
"""
Initializes the QuerySetProxy if not yet initialized.
"""
if not self._check_if_queryset_is_initialized():
self.queryset_proxy.queryset = self._set_queryset()
def _check_if_queryset_is_initialized(self) -> bool:
"""
Checks if the QuerySetProxy is already set and ready.
:return: result of the check
:rtype: bool
"""
return (
hasattr(self.queryset_proxy, "queryset")
and self.queryset_proxy.queryset is not None
)
def _check_if_model_saved(self) -> None:
"""
Verifies if the parent model of the relation has been already saved.
Otherwise QuerySetProxy cannot filter by parent primary key.
"""
pk_value = self._owner.pk
if not pk_value:
raise RelationshipInstanceError(
@ -64,6 +104,14 @@ class RelationProxy(list):
)
def _set_queryset(self) -> "QuerySet":
"""
Creates new QuerySet with relation model and pre filters it with currents
parent model primary key, so all queries by definition are already related
to the parent model only, without need for user to filter them.
:return: initialized QuerySet
:rtype: QuerySet
"""
related_field_name = self.related_field_name
related_field = self.relation.to.Meta.model_fields[related_field_name]
pkname = self._owner.get_column_alias(self._owner.Meta.pkname)
@ -79,6 +127,20 @@ class RelationProxy(list):
async def remove( # type: ignore
self, item: "Model", keep_reversed: bool = True
) -> None:
"""
Removes the item from relation with parent.
Through models are automatically deleted for m2m relations.
For reverse FK relations keep_reversed flag marks if the reversed models
should be kept or deleted from the database too (False means that models
will be deleted, and not only removed from relation).
:param item: child to remove from relation
:type item: Model
:param keep_reversed: flag if the reversed model should be kept or deleted too
:type keep_reversed: bool
"""
if item not in self:
raise NoMatch(
f"Object {self._owner.get_name()} has no "
@ -103,11 +165,19 @@ class RelationProxy(list):
await item.delete()
async def add(self, item: "Model") -> None:
"""
Adds child model to relation.
For ManyToMany relations through instance is automatically created.
:param item: child to add to relation
:type item: Model
"""
relation_name = self.related_field_name
self._check_if_model_saved()
if self.type_ == ormar.RelationType.MULTIPLE:
await self.queryset_proxy.create_through_instance(item)
setattr(item, relation_name, self._owner)
else:
self._check_if_model_saved()
setattr(item, relation_name, self._owner)
await item.update()

19
ormar/relations/utils.py
View File

@ -16,6 +16,25 @@ def get_relations_sides_and_names(
virtual: bool,
relation_name: str,
) -> Tuple["Model", "Model", str, str]:
"""
Determines the names of child and parent relations names, as well as
changes one of the sides of the relation into weakref.proxy to model.
:param to_field: field with relation definition
:type to_field: BaseField
:param parent: parent model
:type parent: Model
:param child: child model
:type child: Model
:param child_name: name of the child
:type child_name: str
:param virtual: flag if relation is virtual
:type virtual: bool
:param relation_name:
:type relation_name:
:return: parent, child, child_name, to_name
:rtype: Tuple["Model", "Model", str, str]
"""
to_name = to_field.name
if issubclass(to_field, ManyToManyField):
child_name = to_field.related_name or child.get_name() + "s"