fix inherited pk and add field accessor access to relations

This commit is contained in:
collerek
2021-06-25 13:32:31 +02:00
parent cc76e9b862
commit 107404c3e8
10 changed files with 516 additions and 61 deletions

View File

@ -0,0 +1,64 @@
import datetime
import uuid
import databases
import pytest
import sqlalchemy
import ormar
from tests.settings import DATABASE_URL
metadata = sqlalchemy.MetaData()
database = databases.Database(DATABASE_URL)
class BaseMeta(ormar.ModelMeta):
database = database
metadata = metadata
class BaseModel(ormar.Model):
class Meta(ormar.ModelMeta):
abstract = True
id: uuid.UUID = ormar.UUID(
primary_key=True, default=uuid.uuid4, uuid_format="string"
)
created_at: datetime.datetime = ormar.DateTime(default=datetime.datetime.utcnow())
updated_at: datetime.datetime = ormar.DateTime(default=datetime.datetime.utcnow())
class Member(BaseModel):
class Meta(BaseMeta):
tablename = "members"
first_name: str = ormar.String(max_length=50)
last_name: str = ormar.String(max_length=50)
@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)
def test_model_structure():
assert "id" in BaseModel.__fields__
assert "id" in BaseModel.Meta.model_fields
assert BaseModel.Meta.model_fields["id"].has_default()
assert BaseModel.__fields__["id"].default_factory is not None
assert "id" in Member.__fields__
assert "id" in Member.Meta.model_fields
assert Member.Meta.model_fields["id"].has_default()
assert Member.__fields__["id"].default_factory is not None
@pytest.mark.asyncio
async def test_fields_inherited_with_default():
async with database:
await Member(first_name="foo", last_name="bar").save()
await Member.objects.create(first_name="foo", last_name="bar")

View File

@ -0,0 +1,100 @@
from typing import List, Optional
import databases
import pytest
import sqlalchemy
import ormar
from tests.settings import DATABASE_URL
database = databases.Database(DATABASE_URL, force_rollback=True)
metadata = sqlalchemy.MetaData()
class Author(ormar.Model):
class Meta:
tablename = "authors"
database = database
metadata = metadata
id: int = ormar.Integer(primary_key=True)
first_name: str = ormar.String(max_length=80)
last_name: str = ormar.String(max_length=80)
class Category(ormar.Model):
class Meta:
tablename = "categories"
database = database
metadata = metadata
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=40)
class Post(ormar.Model):
class Meta:
tablename = "posts"
database = database
metadata = metadata
id: int = ormar.Integer(primary_key=True)
title: str = ormar.String(max_length=200)
categories: Optional[List[Category]] = ormar.ManyToMany(Category)
author: Optional[Author] = ormar.ForeignKey(Author, related_name="author_posts")
@pytest.fixture(autouse=True, scope="module")
def create_test_database():
engine = sqlalchemy.create_engine(DATABASE_URL)
metadata.create_all(engine)
yield
metadata.drop_all(engine)
@pytest.fixture(scope="function")
async def cleanup():
yield
async with database:
PostCategory = Post.Meta.model_fields["categories"].through
await PostCategory.objects.delete(each=True)
await Post.objects.delete(each=True)
await Category.objects.delete(each=True)
await Author.objects.delete(each=True)
@pytest.mark.asyncio
async def test_selecting_related(cleanup):
async with database:
guido = await Author.objects.create(first_name="Guido", last_name="Van Rossum")
post = await Post.objects.create(title="Hello, M2M", author=guido)
news = await Category.objects.create(name="News")
recent = await Category.objects.create(name="Recent")
await post.categories.add(news)
await post.categories.add(recent)
assert len(await post.categories.all()) == 2
# Loads categories and posts (2 queries) and perform the join in Python.
categories = await Category.objects.select_related(Category.posts).all()
assert len(categories) == 2
assert categories[0].name == "News"
news_posts = await news.posts.select_related(Post.author).all()
assert news_posts[0].author == guido
assert (await post.categories.limit(1).all())[0] == news
assert (await post.categories.offset(1).limit(1).all())[0] == recent
assert await post.categories.first() == news
assert await post.categories.exists()
author = await Author.objects.prefetch_related(
Author.author_posts.categories
).get()
assert len(author.author_posts) == 1
assert author.author_posts[0].title == "Hello, M2M"
assert author.author_posts[0].categories[0].name == "News"
assert author.author_posts[0].categories[1].name == "Recent"
post = await Post.objects.select_related([Post.author, Post.categories]).get()
assert len(post.categories) == 2
assert post.categories[0].name == "News"
assert post.categories[1].name == "Recent"
assert post.author.first_name == "Guido"