add tests for cross model forward references, add docs for processing forwardrefs, wip on refactoring queries into separate pages based on functionality

This commit is contained in:
collerek
2021-01-26 17:29:40 +01:00
parent a2834666fc
commit b710ed9780
39 changed files with 2054 additions and 1004 deletions

View File

@ -0,0 +1,114 @@
# type: ignore
import databases
import pytest
import sqlalchemy as sa
from pydantic.typing import ForwardRef
from sqlalchemy import create_engine
import ormar
from ormar import ModelMeta
from tests.settings import DATABASE_URL
metadata = sa.MetaData()
db = databases.Database(DATABASE_URL)
engine = create_engine(DATABASE_URL)
TeacherRef = ForwardRef("Teacher")
class Student(ormar.Model):
class Meta(ModelMeta):
metadata = metadata
database = db
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
primary_teacher: TeacherRef = ormar.ForeignKey(
TeacherRef, related_name="own_students"
)
class StudentTeacher(ormar.Model):
class Meta(ModelMeta):
tablename = "students_x_teachers"
metadata = metadata
database = db
class Teacher(ormar.Model):
class Meta(ModelMeta):
metadata = metadata
database = db
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
students = ormar.ManyToMany(
Student, through=StudentTeacher, related_name="teachers"
)
Student.update_forward_refs()
@pytest.fixture(autouse=True, scope="module")
def create_test_database():
metadata.create_all(engine)
yield
metadata.drop_all(engine)
@pytest.mark.asyncio
async def test_double_relations():
t1 = await Teacher.objects.create(name="Mr. Jones")
t2 = await Teacher.objects.create(name="Ms. Smith")
t3 = await Teacher.objects.create(name="Mr. Quibble")
s1 = await Student.objects.create(name="Joe", primary_teacher=t1)
s2 = await Student.objects.create(name="Sam", primary_teacher=t1)
s3 = await Student.objects.create(name="Kate", primary_teacher=t2)
s4 = await Student.objects.create(name="Zoe", primary_teacher=t2)
s5 = await Student.objects.create(name="John", primary_teacher=t3)
s6 = await Student.objects.create(name="Anna", primary_teacher=t3)
for t in [t1, t2, t3]:
for s in [s1, s2, s3, s4, s5, s6]:
await t.students.add(s)
jones = (
await Teacher.objects.select_related(["students", "own_students"])
.order_by(["students__name", "own_students__name"])
.get(name="Mr. Jones")
)
assert len(jones.students) == 6
assert jones.students[0].name == "Anna"
assert jones.students[5].name == "Zoe"
assert len(jones.own_students) == 2
assert jones.own_students[0].name == "Joe"
assert jones.own_students[1].name == "Sam"
smith = (
await Teacher.objects.select_related(["students", "own_students"])
.filter(students__name__contains="a")
.order_by(["students__name", "own_students__name"])
.get(name="Ms. Smith")
)
assert len(smith.students) == 3
assert smith.students[0].name == "Anna"
assert smith.students[2].name == "Sam"
assert len(smith.own_students) == 2
assert smith.own_students[0].name == "Kate"
assert smith.own_students[1].name == "Zoe"
quibble = (
await Teacher.objects.select_related(["students", "own_students"])
.filter(students__name__startswith="J")
.order_by(["-students__name", "own_students__name"])
.get(name="Mr. Quibble")
)
assert len(quibble.students) == 2
assert quibble.students[1].name == "Joe"
assert quibble.students[0].name == "John"
assert len(quibble.own_students) == 2
assert quibble.own_students[1].name == "John"
assert quibble.own_students[0].name == "Anna"

View File

@ -1,4 +1,6 @@
# type: ignore
from typing import List
import databases
import pytest
import sqlalchemy
@ -15,7 +17,7 @@ metadata = sa.MetaData()
db = databases.Database(DATABASE_URL)
engine = create_engine(DATABASE_URL)
Person = ForwardRef("Person")
PersonRef = ForwardRef("Person")
class Person(ormar.Model):
@ -25,19 +27,14 @@ class Person(ormar.Model):
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
supervisor: Person = ormar.ForeignKey(Person, related_name="employees")
supervisor: PersonRef = ormar.ForeignKey(PersonRef, related_name="employees")
Person.update_forward_refs()
Game = ForwardRef("Game")
Child = ForwardRef("Child")
class ChildFriend(ormar.Model):
class Meta(ModelMeta):
metadata = metadata
database = db
GameRef = ForwardRef("Game")
ChildRef = ForwardRef("Child")
ChildFriendRef = ForwardRef("ChildFriend")
class Child(ormar.Model):
@ -47,9 +44,19 @@ class Child(ormar.Model):
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
favourite_game: Game = ormar.ForeignKey(Game, related_name="liked_by")
least_favourite_game: Game = ormar.ForeignKey(Game, related_name="not_liked_by")
friends = ormar.ManyToMany(Child, through=ChildFriend, related_name="also_friends")
favourite_game: GameRef = ormar.ForeignKey(GameRef, related_name="liked_by")
least_favourite_game: GameRef = ormar.ForeignKey(
GameRef, related_name="not_liked_by"
)
friends = ormar.ManyToMany(
ChildRef, through=ChildFriendRef, related_name="also_friends"
)
class ChildFriend(ormar.Model):
class Meta(ModelMeta):
metadata = metadata
database = db
class Game(ormar.Model):
@ -82,8 +89,8 @@ async def cleanup():
@pytest.mark.asyncio
async def test_not_uprated_model_raises_errors():
Person2 = ForwardRef("Person2")
async def test_not_updated_model_raises_errors():
Person2Ref = ForwardRef("Person2")
class Person2(ormar.Model):
class Meta(ModelMeta):
@ -92,7 +99,7 @@ async def test_not_uprated_model_raises_errors():
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
supervisor: Person2 = ormar.ForeignKey(Person2, related_name="employees")
supervisor: Person2Ref = ormar.ForeignKey(Person2Ref, related_name="employees")
with pytest.raises(ModelError):
await Person2.objects.create(name="Test")
@ -104,6 +111,74 @@ async def test_not_uprated_model_raises_errors():
await Person2.objects.get()
@pytest.mark.asyncio
async def test_not_updated_model_m2m_raises_errors():
Person3Ref = ForwardRef("Person3")
class PersonFriend(ormar.Model):
class Meta(ModelMeta):
metadata = metadata
database = db
class Person3(ormar.Model):
class Meta(ModelMeta):
metadata = metadata
database = db
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
supervisors: Person3Ref = ormar.ManyToMany(
Person3Ref, through=PersonFriend, related_name="employees"
)
with pytest.raises(ModelError):
await Person3.objects.create(name="Test")
with pytest.raises(ModelError):
Person3(name="Test")
with pytest.raises(ModelError):
await Person3.objects.get()
@pytest.mark.asyncio
async def test_not_updated_model_m2m_through_raises_errors():
PersonPetRef = ForwardRef("PersonPet")
class Pet(ormar.Model):
class Meta(ModelMeta):
metadata = metadata
database = db
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
class Person4(ormar.Model):
class Meta(ModelMeta):
metadata = metadata
database = db
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
pets: List[Pet] = ormar.ManyToMany(
Pet, through=PersonPetRef, related_name="owners"
)
class PersonPet(ormar.Model):
class Meta(ModelMeta):
metadata = metadata
database = db
with pytest.raises(ModelError):
await Person4.objects.create(name="Test")
with pytest.raises(ModelError):
Person4(name="Test")
with pytest.raises(ModelError):
await Person4.objects.get()
def test_proper_field_init():
assert "supervisor" in Person.Meta.model_fields
assert Person.Meta.model_fields["supervisor"].to == Person