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

ugly but working - to refactor

Browse Source
This commit is contained in:
collerek
2021-03-11 17:53:13 +01:00
parent e306eecc2c
commit 0ae340100e
16 changed files with 688 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
ormar/fields/base.py
View File

@ -54,6 +54,8 @@ class BaseField(FieldInfo):
through: Type["Model"]
self_reference: bool = False
self_reference_primary: Optional[str] = None
orders_by: Optional[List[str]] = None
related_orders_by: Optional[List[str]] = None
encrypt_secret: str
encrypt_backend: EncryptBackends = EncryptBackends.NONE

53
ormar/fields/foreign_key.py
View File

@ -3,7 +3,7 @@ import sys
import uuid
from dataclasses import dataclass
from random import choices
from typing import Any, List, Optional, TYPE_CHECKING, Tuple, Type, Union
from typing import Any, Dict, List, Optional, TYPE_CHECKING, Tuple, Type, Union
import sqlalchemy
from pydantic import BaseModel, create_model
@ -119,6 +119,35 @@ def populate_fk_params_based_on_to_model(
return __type__, constraints, column_type
def validate_not_allowed_fields(kwargs: Dict) -> None:
"""
Verifies if not allowed parameters are set on relation models.
Usually they are omitted later anyway but this way it's explicitly
notify the user that it's not allowed/ supported.
:raises ModelDefinitionError: if any forbidden field is set
:param kwargs: dict of kwargs to verify passed to relation field
:type kwargs: Dict
"""
default = kwargs.pop("default", None)
encrypt_secret = kwargs.pop("encrypt_secret", None)
encrypt_backend = kwargs.pop("encrypt_backend", None)
encrypt_custom_backend = kwargs.pop("encrypt_custom_backend", None)
not_supported = [
default,
encrypt_secret,
encrypt_backend,
encrypt_custom_backend,
]
if any(x is not None for x in not_supported):
raise ModelDefinitionError(
f"Argument {next((x for x in not_supported if x is not None))} "
f"is not supported "
"on relation fields!"
)
class UniqueColumns(UniqueConstraint):
"""
Subclass of sqlalchemy.UniqueConstraint.
@ -184,24 +213,10 @@ def ForeignKey( # noqa CFQ002
owner = kwargs.pop("owner", None)
self_reference = kwargs.pop("self_reference", False)
orders_by = kwargs.pop("orders_by", None)
related_orders_by = kwargs.pop("related_orders_by", None)
default = kwargs.pop("default", None)
encrypt_secret = kwargs.pop("encrypt_secret", None)
encrypt_backend = kwargs.pop("encrypt_backend", None)
encrypt_custom_backend = kwargs.pop("encrypt_custom_backend", None)
not_supported = [
default,
encrypt_secret,
encrypt_backend,
encrypt_custom_backend,
]
if any(x is not None for x in not_supported):
raise ModelDefinitionError(
f"Argument {next((x for x in not_supported if x is not None))} "
f"is not supported "
"on relation fields!"
)
validate_not_allowed_fields(kwargs)
if to.__class__ == ForwardRef:
__type__ = to if not nullable else Optional[to]
@ -237,6 +252,8 @@ def ForeignKey( # noqa CFQ002
owner=owner,
self_reference=self_reference,
is_relation=True,
orders_by=orders_by,
related_orders_by=related_orders_by,
)
return type("ForeignKey", (ForeignKeyField, BaseField), namespace)

25
ormar/fields/many_to_many.py
View File

@ -5,7 +5,7 @@ from pydantic.typing import ForwardRef, evaluate_forwardref
import ormar # noqa: I100
from ormar import ModelDefinitionError
from ormar.fields import BaseField
from ormar.fields.foreign_key import ForeignKeyField
from ormar.fields.foreign_key import ForeignKeyField, validate_not_allowed_fields
if TYPE_CHECKING: # pragma no cover
from ormar.models import Model
@ -93,26 +93,13 @@ def ManyToMany(
nullable = kwargs.pop("nullable", True)
owner = kwargs.pop("owner", None)
self_reference = kwargs.pop("self_reference", False)
orders_by = kwargs.pop("orders_by", None)
related_orders_by = kwargs.pop("related_orders_by", None)
if through is not None and through.__class__ != ForwardRef:
forbid_through_relations(cast(Type["Model"], through))
default = kwargs.pop("default", None)
encrypt_secret = kwargs.pop("encrypt_secret", None)
encrypt_backend = kwargs.pop("encrypt_backend", None)
encrypt_custom_backend = kwargs.pop("encrypt_custom_backend", None)
not_supported = [
default,
encrypt_secret,
encrypt_backend,
encrypt_custom_backend,
]
if any(x is not None for x in not_supported):
raise ModelDefinitionError(
f"Argument {next((x for x in not_supported if x is not None))} "
f"is not supported "
"on relation fields!"
)
validate_not_allowed_fields(kwargs)
if to.__class__ == ForwardRef:
__type__ = to if not nullable else Optional[to]
@ -141,6 +128,8 @@ def ManyToMany(
self_reference=self_reference,
is_relation=True,
is_multi=True,
orders_by=orders_by,
related_orders_by=related_orders_by,
)
return type("ManyToMany", (ManyToManyField, BaseField), namespace)

2
ormar/models/helpers/models.py
View File

@ -51,6 +51,8 @@ def populate_default_options_values(
new_model.Meta.model_fields = model_fields
if not hasattr(new_model.Meta, "abstract"):
new_model.Meta.abstract = False
if not hasattr(new_model.Meta, "order_by"):
new_model.Meta.order_by = []
if any(
is_field_an_forward_ref(field) for field in new_model.Meta.model_fields.values()

2
ormar/models/helpers/relations.py
View File

@ -110,6 +110,7 @@ def register_reverse_model_fields(model_field: Type["ForeignKeyField"]) -> None:
owner=model_field.to,
self_reference=model_field.self_reference,
self_reference_primary=model_field.self_reference_primary,
orders_by=model_field.related_orders_by,
)
# register foreign keys on through model
model_field = cast(Type["ManyToManyField"], model_field)
@ -123,6 +124,7 @@ def register_reverse_model_fields(model_field: Type["ForeignKeyField"]) -> None:
related_name=model_field.name,
owner=model_field.to,
self_reference=model_field.self_reference,
orders_by=model_field.related_orders_by,
)

3
ormar/models/helpers/sqlalchemy.py
View File

@ -252,6 +252,9 @@ def populate_meta_tablename_columns_and_pk(
new_model.Meta.columns = columns
new_model.Meta.pkname = pkname
if not new_model.Meta.order_by:
# by default we sort by pk name if other option not provided
new_model.Meta.order_by.append(pkname)
return new_model

1
ormar/models/metaclass.py
View File

@ -71,6 +71,7 @@ class ModelMeta:
signals: SignalEmitter
abstract: bool
requires_ref_update: bool
order_by: List[str]
def add_cached_properties(new_model: Type["Model"]) -> None:

9
ormar/models/model.py
View File

@ -273,7 +273,10 @@ class Model(ModelRow):
return self
async def load_all(
self: T, follow: bool = False, exclude: Union[List, str, Set, Dict] = None
self: T,
follow: bool = False,
exclude: Union[List, str, Set, Dict] = None,
order_by: Union[List, str] = None,
) -> T:
"""
Allow to refresh existing Models fields from database.
@ -291,6 +294,8 @@ class Model(ModelRow):
will load second Model A but will never follow into Model X.
Nested relations of those kind need to be loaded manually.
:param order_by: columns by which models should be sorted
:type order_by: Union[List, str]
:raises NoMatch: If given pk is not found in database.
:param exclude: related models to exclude
@ -308,6 +313,8 @@ class Model(ModelRow):
queryset = self.__class__.objects
if exclude:
queryset = queryset.exclude_fields(exclude)
if order_by:
queryset = queryset.order_by(order_by)
instance = await queryset.select_related(relations).get(pk=self.pk)
self._orm.clear()
self.update_from_dict(instance.dict())

115
ormar/queryset/join.py
View File

@ -1,49 +1,54 @@
from collections import OrderedDict
from typing import (
Any,
Dict,
List,
Optional,
TYPE_CHECKING,
Tuple,
Type,
Type, cast,
)
import sqlalchemy
from sqlalchemy import text
import ormar # noqa I100
from ormar.exceptions import RelationshipInstanceError
from ormar.exceptions import ModelDefinitionError, RelationshipInstanceError
from ormar.relations import AliasManager
if TYPE_CHECKING: # pragma no cover
from ormar import Model
from ormar import Model, ManyToManyField
from ormar.queryset import OrderAction
from ormar.models.excludable import ExcludableItems
class SqlJoin:
def __init__( # noqa: CFQ002
self,
used_aliases: List,
select_from: sqlalchemy.sql.select,
columns: List[sqlalchemy.Column],
excludable: "ExcludableItems",
order_columns: Optional[List["OrderAction"]],
sorted_orders: OrderedDict,
main_model: Type["Model"],
relation_name: str,
relation_str: str,
related_models: Any = None,
own_alias: str = "",
source_model: Type["Model"] = None,
self,
used_aliases: List,
select_from: sqlalchemy.sql.select,
columns: List[sqlalchemy.Column],
excludable: "ExcludableItems",
order_columns: Optional[List["OrderAction"]],
sorted_orders: OrderedDict,
main_model: Type["Model"],
relation_name: str,
relation_str: str,
related_models: Any = None,
own_alias: str = "",
source_model: Type["Model"] = None,
already_sorted: Dict = None,
) -> None:
self.relation_name = relation_name
self.related_models = related_models or []
self.select_from = select_from
self.columns = columns
self.excludable = excludable
self.order_columns = order_columns
self.sorted_orders = sorted_orders
self.already_sorted = already_sorted or dict()
self.main_model = main_model
self.own_alias = own_alias
self.used_aliases = used_aliases
@ -97,7 +102,7 @@ class SqlJoin:
return self.next_model.Meta.table
def _on_clause(
self, previous_alias: str, from_clause: str, to_clause: str,
self, previous_alias: str, from_clause: str, to_clause: str,
) -> text:
"""
Receives aliases and names of both ends of the join and combines them
@ -169,8 +174,8 @@ class SqlJoin:
for related_name in self.related_models:
remainder = None
if (
isinstance(self.related_models, dict)
and self.related_models[related_name]
isinstance(self.related_models, dict)
and self.related_models[related_name]
):
remainder = self.related_models[related_name]
self._process_deeper_join(related_name=related_name, remainder=remainder)
@ -205,6 +210,7 @@ class SqlJoin:
relation_str="__".join([self.relation_str, related_name]),
own_alias=self.next_alias,
source_model=self.source_model or self.main_model,
already_sorted=self.already_sorted,
)
(
self.used_aliases,
@ -251,18 +257,18 @@ class SqlJoin:
"""
target_field = self.target_field
is_primary_self_ref = (
target_field.self_reference
and self.relation_name == target_field.self_reference_primary
target_field.self_reference
and self.relation_name == target_field.self_reference_primary
)
if (is_primary_self_ref and not reverse) or (
not is_primary_self_ref and reverse
not is_primary_self_ref and reverse
):
new_part = target_field.default_source_field_name() # type: ignore
else:
new_part = target_field.default_target_field_name() # type: ignore
return new_part
def _process_join(self,) -> None: # noqa: CFQ002
def _process_join(self, ) -> None: # noqa: CFQ002
"""
Resolves to and from column names and table names.
@ -307,12 +313,11 @@ class SqlJoin:
self.used_aliases.append(self.next_alias)
def _set_default_primary_key_order_by(self) -> None:
clause = ormar.OrderAction(
order_str=self.next_model.Meta.pkname,
model_cls=self.next_model,
alias=self.next_alias,
)
self.sorted_orders[clause] = clause.get_text_clause()
for order_by in self.next_model.Meta.order_by:
clause = ormar.OrderAction(
order_str=order_by, model_cls=self.next_model, alias=self.next_alias,
)
self.sorted_orders[clause] = clause.get_text_clause()
def _get_order_bys(self) -> None: # noqa: CCR001
"""
@ -320,18 +325,60 @@ class SqlJoin:
Otherwise by default each table is sorted by a primary key column asc.
"""
alias = self.next_alias
current_table_sorted = False
if f"{alias}_{self.next_model.get_name()}" in self.already_sorted:
current_table_sorted = True
if self.order_columns:
current_table_sorted = False
for condition in self.order_columns:
if condition.check_if_filter_apply(
target_model=self.next_model, alias=alias
target_model=self.next_model, alias=alias
):
current_table_sorted = True
self.sorted_orders[condition] = condition.get_text_clause()
if not current_table_sorted and not self.target_field.is_multi:
self._set_default_primary_key_order_by()
self.already_sorted[
f"{self.next_alias}_{self.next_model.get_name()}"
] = condition
# TODO: refactor into smaller helper functions
if self.target_field.orders_by and not current_table_sorted:
current_table_sorted = True
for order_by in self.target_field.orders_by:
if self.target_field.is_multi and "__" in order_by:
parts = order_by.split("__")
if (
len(parts) > 2
or parts[0] != self.target_field.through.get_name()
):
raise ModelDefinitionError(
"You can order the relation only"
"by related or link table columns!"
)
model = self.target_field.owner
clause = ormar.OrderAction(
order_str=order_by, model_cls=model, alias=alias,
)
elif self.target_field.is_multi:
alias = self.alias_manager.resolve_relation_alias(
from_model=self.target_field.through,
relation_name=cast("ManyToManyField",
self.target_field).default_target_field_name(),
)
model = self.target_field.to
clause = ormar.OrderAction(
order_str=order_by, model_cls=model, alias=alias
)
else:
alias = self.alias_manager.resolve_relation_alias(
from_model=self.target_field.owner,
relation_name=self.target_field.name,
)
model = self.target_field.to
clause = ormar.OrderAction(
order_str=order_by, model_cls=model, alias=alias
)
self.sorted_orders[clause] = clause.get_text_clause()
self.already_sorted[f"{alias}_{model.get_name()}"] = clause
elif not self.target_field.is_multi:
if not current_table_sorted and not self.target_field.is_multi:
self._set_default_primary_key_order_by()
def _get_to_and_from_keys(self) -> Tuple[str, str]:

12
ormar/queryset/query.py
View File

@ -63,15 +63,17 @@ class Query:
That way the subquery with limit and offset only on main model has proper
sorting applied and correct models are fetched.
"""
current_table_sorted = False
if self.order_columns:
for clause in self.order_columns:
if clause.is_source_model_order:
current_table_sorted = True
self.sorted_orders[clause] = clause.get_text_clause()
else:
clause = ormar.OrderAction(
order_str=self.model_cls.Meta.pkname, model_cls=self.model_cls
)
self.sorted_orders[clause] = clause.get_text_clause()
if not current_table_sorted:
for order_by in self.model_cls.Meta.order_by:
clause = ormar.OrderAction(order_str=order_by, model_cls=self.model_cls)
self.sorted_orders[clause] = clause.get_text_clause()
def _pagination_query_required(self) -> bool:
"""

1
tests/test_columns.py
View File

@ -1,5 +1,4 @@
import datetime
import os
import databases
import pydantic

113
tests/test_default_model_order.py Normal file
View File

@ -0,0 +1,113 @@
from typing import Optional
import databases
import pytest
import sqlalchemy
import ormar
from tests.settings import DATABASE_URL
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()
class BaseMeta(ormar.ModelMeta):
metadata = metadata
database = database
class Author(ormar.Model):
class Meta(BaseMeta):
tablename = "authors"
order_by = ["-name"]
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
class Book(ormar.Model):
class Meta(BaseMeta):
tablename = "books"
order_by = ["year", "-ranking"]
id: int = ormar.Integer(primary_key=True)
author: Optional[Author] = ormar.ForeignKey(Author)
title: str = ormar.String(max_length=100)
year: int = ormar.Integer(nullable=True)
ranking: int = ormar.Integer(nullable=True)
@pytest.fixture(autouse=True, scope="module")
def create_test_database():
engine = sqlalchemy.create_engine(DATABASE_URL)
metadata.drop_all(engine)
metadata.create_all(engine)
yield
metadata.drop_all(engine)
@pytest.fixture(autouse=True, scope="function")
async def cleanup():
await Book.objects.delete(each=True)
await Author.objects.delete(each=True)
@pytest.mark.asyncio
async def test_default_orders_is_applied():
async with database:
tolkien = await Author(name="J.R.R. Tolkien").save()
sapkowski = await Author(name="Andrzej Sapkowski").save()
king = await Author(name="Stephen King").save()
lewis = await Author(name="C.S Lewis").save()
authors = await Author.objects.all()
assert authors[0] == king
assert authors[1] == tolkien
assert authors[2] == lewis
assert authors[3] == sapkowski
authors = await Author.objects.order_by("name").all()
assert authors[3] == king
assert authors[2] == tolkien
assert authors[1] == lewis
assert authors[0] == sapkowski
@pytest.mark.asyncio
async def test_default_orders_is_applied_on_related():
async with database:
tolkien = await Author(name="J.R.R. Tolkien").save()
silmarillion = await Book(
author=tolkien, title="The Silmarillion", year=1977
).save()
lotr = await Book(
author=tolkien, title="The Lord of the Rings", year=1955
).save()
hobbit = await Book(author=tolkien, title="The Hobbit", year=1933).save()
await tolkien.books.all()
assert tolkien.books[0] == hobbit
assert tolkien.books[1] == lotr
assert tolkien.books[2] == silmarillion
await tolkien.books.order_by("-title").all()
assert tolkien.books[2] == hobbit
assert tolkien.books[1] == lotr
assert tolkien.books[0] == silmarillion
@pytest.mark.asyncio
async def test_default_orders_is_applied_on_related_two_fields():
async with database:
sanders = await Author(name="Brandon Sanderson").save()
twok = await Book(
author=sanders, title="The Way of Kings", year=2010, ranking=10
).save()
bret = await Author(name="Peter V. Bret").save()
tds = await Book(
author=bret, title="The Desert Spear", year=2010, ranking=9
).save()
books = await Book.objects.all()
assert books[0] == twok
assert books[1] == tds

149
tests/test_default_relation_order.py Normal file
View File

@ -0,0 +1,149 @@
from typing import List, Optional
from uuid import UUID, uuid4
import databases
import pytest
import sqlalchemy
import ormar
from tests.settings import DATABASE_URL
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()
class BaseMeta(ormar.ModelMeta):
metadata = metadata
database = database
class Author(ormar.Model):
class Meta(BaseMeta):
tablename = "authors"
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
class Book(ormar.Model):
class Meta(BaseMeta):
tablename = "books"
id: int = ormar.Integer(primary_key=True)
author: Optional[Author] = ormar.ForeignKey(
Author, orders_by=["name"], related_orders_by=["-year"]
)
title: str = ormar.String(max_length=100)
year: int = ormar.Integer(nullable=True)
ranking: int = ormar.Integer(nullable=True)
class Animal(ormar.Model):
class Meta(BaseMeta):
tablename = "animals"
id: UUID = ormar.UUID(primary_key=True, default=uuid4)
name: str = ormar.String(max_length=200)
specie: str = ormar.String(max_length=200)
class Human(ormar.Model):
class Meta(BaseMeta):
tablename = "humans"
id: UUID = ormar.UUID(primary_key=True, default=uuid4)
name: str = ormar.Text(default="")
pets: List[Animal] = ormar.ManyToMany(
Animal,
related_name="care_takers",
orders_by=["specie", "-name"],
related_orders_by=["name"],
)
@pytest.fixture(autouse=True, scope="module")
def create_test_database():
engine = sqlalchemy.create_engine(DATABASE_URL)
metadata.drop_all(engine)
metadata.create_all(engine)
yield
metadata.drop_all(engine)
@pytest.fixture(autouse=True, scope="function")
async def cleanup():
await Book.objects.delete(each=True)
await Author.objects.delete(each=True)
@pytest.mark.asyncio
async def test_default_orders_is_applied_from_reverse_relation():
async with database:
tolkien = await Author(name="J.R.R. Tolkien").save()
hobbit = await Book(author=tolkien, title="The Hobbit", year=1933).save()
silmarillion = await Book(
author=tolkien, title="The Silmarillion", year=1977
).save()
lotr = await Book(
author=tolkien, title="The Lord of the Rings", year=1955
).save()
tolkien = await Author.objects.select_related("books").get()
assert tolkien.books[2] == hobbit
assert tolkien.books[1] == lotr
assert tolkien.books[0] == silmarillion
@pytest.mark.asyncio
async def test_default_orders_is_applied_from_relation():
async with database:
bret = await Author(name="Peter V. Bret").save()
tds = await Book(
author=bret, title="The Desert Spear", year=2010, ranking=9
).save()
sanders = await Author(name="Brandon Sanderson").save()
twok = await Book(
author=sanders, title="The Way of Kings", year=2010, ranking=10
).save()
books = await Book.objects.order_by("year").select_related("author").all()
assert books[0] == twok
assert books[1] == tds
@pytest.mark.asyncio
async def test_default_orders_is_applied_from_relation_on_m2m():
async with database:
alice = await Human(name="Alice").save()
spot = await Animal(name="Spot", specie="Cat").save()
zkitty = await Animal(name="ZKitty", specie="Cat").save()
noodle = await Animal(name="Noodle", specie="Anaconda").save()
await alice.pets.add(noodle)
await alice.pets.add(spot)
await alice.pets.add(zkitty)
await alice.load_all()
assert alice.pets[0] == noodle
assert alice.pets[1] == zkitty
assert alice.pets[2] == spot
@pytest.mark.asyncio
async def test_default_orders_is_applied_from_reverse_relation_on_m2m():
async with database:
max = await Animal(name="Max", specie="Dog").save()
joe = await Human(name="Joe").save()
zack = await Human(name="Zack").save()
julia = await Human(name="Julia").save()
await max.care_takers.add(joe)
await max.care_takers.add(zack)
await max.care_takers.add(julia)
await max.load_all()
assert max.care_takers[0] == joe
assert max.care_takers[1] == julia
assert max.care_takers[2] == zack

176
tests/test_default_through_relation_order.py Normal file
View File

@ -0,0 +1,176 @@
from typing import List
from uuid import UUID, uuid4
import databases
import pytest
import sqlalchemy
import ormar
from tests.settings import DATABASE_URL
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()
class BaseMeta(ormar.ModelMeta):
metadata = metadata
database = database
class Animal(ormar.Model):
class Meta(BaseMeta):
tablename = "animals"
id: UUID = ormar.UUID(primary_key=True, default=uuid4)
name: str = ormar.Text(default="")
# favoriteHumans
class Link(ormar.Model):
class Meta(BaseMeta):
tablename = "link_table"
id: int = ormar.Integer(primary_key=True)
animal_order: int = ormar.Integer(nullable=True)
human_order: int = ormar.Integer(nullable=True)
class Human(ormar.Model):
class Meta(BaseMeta):
tablename = "humans"
id: UUID = ormar.UUID(primary_key=True, default=uuid4)
name: str = ormar.Text(default="")
favoriteAnimals: List[Animal] = ormar.ManyToMany(
Animal,
through=Link,
related_name="favoriteHumans",
orders_by=["link__animal_order"],
related_orders_by=["link__human_order"],
)
@pytest.fixture(autouse=True, scope="module")
def create_test_database():
engine = sqlalchemy.create_engine(DATABASE_URL)
metadata.drop_all(engine)
metadata.create_all(engine)
yield
metadata.drop_all(engine)
@pytest.mark.asyncio
async def test_ordering_by_through_on_m2m_field():
async with database:
alice = await Human(name="Alice").save()
bob = await Human(name="Bob").save()
charlie = await Human(name="Charlie").save()
spot = await Animal(name="Spot").save()
kitty = await Animal(name="Kitty").save()
noodle = await Animal(name="Noodle").save()
# you need to add them in order anyway so can provide order explicitly
# if you have a lot of them a list with enumerate might be an option
await alice.favoriteAnimals.add(noodle, animal_order=0, human_order=0)
await alice.favoriteAnimals.add(spot, animal_order=1, human_order=0)
await alice.favoriteAnimals.add(kitty, animal_order=2, human_order=0)
# you dont have to reload queries on queryset clears the existing related
# alice = await alice.reload()
await alice.load_all()
assert [x.name for x in alice.favoriteAnimals] == ["Noodle", "Spot", "Kitty"]
await bob.favoriteAnimals.add(noodle, animal_order=0, human_order=1)
await bob.favoriteAnimals.add(kitty, animal_order=1, human_order=1)
await bob.favoriteAnimals.add(spot, animal_order=2, human_order=1)
await bob.load_all()
assert [x.name for x in bob.favoriteAnimals] == ["Noodle", "Kitty", "Spot"]
await charlie.favoriteAnimals.add(kitty, animal_order=0, human_order=2)
await charlie.favoriteAnimals.add(noodle, animal_order=1, human_order=2)
await charlie.favoriteAnimals.add(spot, animal_order=2, human_order=2)
await charlie.load_all()
assert [x.name for x in charlie.favoriteAnimals] == ["Kitty", "Noodle", "Spot"]
animals = [noodle, kitty, spot]
for animal in animals:
await animal.load_all()
assert [x.name for x in animal.favoriteHumans] == [
"Alice",
"Bob",
"Charlie",
]
zack = await Human(name="Zack").save()
async def reorder_humans(animal, new_ordered_humans):
noodle_links = await Link.objects.filter(animal=animal).all()
for link in noodle_links:
link.human_order = next(
(
i
for i, x in enumerate(new_ordered_humans)
if x.pk == link.human.pk
),
None,
)
await Link.objects.bulk_update(noodle_links, columns=["human_order"])
await noodle.favoriteHumans.add(zack, animal_order=0, human_order=0)
await reorder_humans(noodle, [zack, alice, bob, charlie])
await noodle.load_all()
assert [x.name for x in noodle.favoriteHumans] == [
"Zack",
"Alice",
"Bob",
"Charlie",
]
await zack.load_all()
assert [x.name for x in zack.favoriteAnimals] == ["Noodle"]
humans = noodle.favoriteHumans
humans.insert(1, humans.pop(0))
await reorder_humans(noodle, humans)
await noodle.load_all()
assert [x.name for x in noodle.favoriteHumans] == [
"Alice",
"Zack",
"Bob",
"Charlie",
]
humans.insert(2, humans.pop(1))
await reorder_humans(noodle, humans)
await noodle.load_all()
assert [x.name for x in noodle.favoriteHumans] == [
"Alice",
"Bob",
"Zack",
"Charlie",
]
humans.insert(3, humans.pop(2))
await reorder_humans(noodle, humans)
await noodle.load_all()
assert [x.name for x in noodle.favoriteHumans] == [
"Alice",
"Bob",
"Charlie",
"Zack",
]
await kitty.favoriteHumans.remove(bob)
await kitty.load_all()
assert [x.name for x in kitty.favoriteHumans] == ["Alice", "Charlie"]
bob = await noodle.favoriteHumans.get(pk=bob.pk)
assert bob.link.human_order == 1
await noodle.favoriteHumans.remove(
await noodle.favoriteHumans.filter(link__human_order=2).get()
)
await noodle.load_all()
assert [x.name for x in noodle.favoriteHumans] == ["Alice", "Bob", "Zack"]

24
tests/test_load_all.py
View File

@ -107,6 +107,30 @@ async def test_load_all_many_to_many():
assert hq.nicks[1].name == "Bazinga20"
@pytest.mark.asyncio
async def test_load_all_with_order():
async with database:
async with database.transaction(force_rollback=True):
nick1 = await NickName.objects.create(name="Barry", is_lame=False)
nick2 = await NickName.objects.create(name="Joe", is_lame=True)
hq = await HQ.objects.create(name="Main")
await hq.nicks.add(nick1)
await hq.nicks.add(nick2)
hq = await HQ.objects.get(name="Main")
await hq.load_all(order_by="-nicks__name")
assert hq.nicks[0] == nick2
assert hq.nicks[0].name == "Joe"
assert hq.nicks[1] == nick1
assert hq.nicks[1].name == "Barry"
await hq.load_all()
assert hq.nicks[0] == nick1
assert hq.nicks[1] == nick2
@pytest.mark.asyncio
async def test_loading_reversed_relation():
async with database:

78
tests/test_proper_order_of_sorting_apply.py Normal file
View File

@ -0,0 +1,78 @@
from typing import Optional
import databases
import pytest
import sqlalchemy
import ormar
from tests.settings import DATABASE_URL
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()
class BaseMeta(ormar.ModelMeta):
metadata = metadata
database = database
class Author(ormar.Model):
class Meta(BaseMeta):
tablename = "authors"
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
class Book(ormar.Model):
class Meta(BaseMeta):
tablename = "books"
order_by = ["-ranking"]
id: int = ormar.Integer(primary_key=True)
author: Optional[Author] = ormar.ForeignKey(
Author, orders_by=["name"], related_orders_by=["-year"]
)
title: str = ormar.String(max_length=100)
year: int = ormar.Integer(nullable=True)
ranking: int = ormar.Integer(nullable=True)
@pytest.fixture(autouse=True, scope="module")
def create_test_database():
engine = sqlalchemy.create_engine(DATABASE_URL)
metadata.drop_all(engine)
metadata.create_all(engine)
yield
metadata.drop_all(engine)
@pytest.fixture(autouse=True, scope="function")
async def cleanup():
await Book.objects.delete(each=True)
await Author.objects.delete(each=True)
@pytest.mark.asyncio
async def test_default_orders_is_applied_from_reverse_relation():
async with database:
tolkien = await Author(name="J.R.R. Tolkien").save()
hobbit = await Book(author=tolkien, title="The Hobbit", year=1933).save()
silmarillion = await Book(
author=tolkien, title="The Silmarillion", year=1977
).save()
lotr = await Book(
author=tolkien, title="The Lord of the Rings", year=1955
).save()
tolkien = await Author.objects.select_related("books").get()
assert tolkien.books[2] == hobbit
assert tolkien.books[1] == lotr
assert tolkien.books[0] == silmarillion
tolkien = (
await Author.objects.select_related("books").order_by("books__title").get()
)
assert tolkien.books[0] == hobbit
assert tolkien.books[1] == lotr
assert tolkien.books[2] == silmarillion