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

WIP - Pydantic v2 support (#1238)

Browse Source 
* WIP

* WIP - make test_model_definition tests pass

* WIP - make test_model_methods pass

* WIP - make whole test suit at least run - failing 49/443 tests

* WIP fix part of the getting pydantic tests as types of fields are now kept in core schema and not on fieldsinfo

* WIP fix validation in update by creating individual fields validators, failing 36/443

* WIP fix __pydantic_extra__ in intializing model, fix test related to pydantic config checks, failing 32/442

* WIP - fix enum schema in model_json_schema, failing 31/442

* WIP - fix copying through model, fix setting pydantic fields on through, fix default config and inheriting from it, failing 26/442

* WIP fix tests checking pydantic schema, fix excluding parent fields, failing 21/442

* WIP some missed files

* WIP - fix validators inheritance and fix validators in generated pydantic, failing 17/442

* WIP - fix through models setting - only on reverse side of relation, but always on reverse side, failing 15/442

* WIP - fix through models setting - only on reverse side of relation, but always on reverse side, failing 15/442

* WIP - working on proper populating __dict__ for relations for new schema dumping, some work on openapi docs, failing 13/442

* WIP - remove property fields as pydantic has now computed_field on its own, failing 9/442

* WIP - fixes in docs, failing 8/442

* WIP - fix tests for largebinary schema, wrapped bytes fields fail in pydantic, will be fixed in pydantic-core, remaining is circural schema for related models, failing 6/442

* WIP - fix to pk only models in schemas

* Getting test suites to pass (#1249)

* wip, fixing tests

* iteration, fixing some more tests

* iteration, fixing some more tests

* adhere to comments

* adhere to comments

* remove unnecessary dict call, re-add getattribute for testing

* todo for reverse relationship

* adhere to comments, remove prints

* solve circular refs

* all tests pass 🎉

* remove 3.7 from tests

* add lint and type check jobs

* reforat with ruff, fix jobs

* rename jobs

* fix imports

* fix evaluate in py3.8

* partially fix coverage

* fix coverage, add more tests

* fix test ids

* fix test ids

* fix lint, fix docs, make docs fully working scripts, add test docs job

* fix pyproject

* pin py ver in test docs

* change dir in test docs

* fix pydantic warning hack

* rm poetry call in test_docs

* switch to pathlib in test docs

* remove coverage req test docs

* fix type check tests, fix part of types

* fix/skip next part of types

* fix next part of types

* fix next part of types

* fix coverage

* fix coverage

* fix type (bit dirty 🤷)

* fix some code smells

* change pre-commit

* tweak workflows

* remove no root from tests

* switch to full python path by passing sys.executable

* some small refactor in new base model, one sample test, change makefile

* small refactors to reduce complexity of methods

* temp add tests for prs against pydantic_v2

* remove all references to __fields__

* remove all references to construct, deprecate the method and update model_construct to be in line with pydantic

* deprecate dict and add model_dump, todo switch to model_dict in calls

* fix tests

* change to union

* change to union

* change to model_dump and model_dump_json from dict and json deprecated methods, deprecate them in ormar too

* finish switching dict() -> model_dump()

* finish switching json() -> model_dump_json()

* remove fully pydantic_only

* switch to extra for payment card, change missed json calls

* fix coverage - no more warnings internal

* fix coverage - no more warnings internal - part 2

* split model_construct into own and pydantic parts

* split determine pydantic field type

* change to new field validators

* fix benchmarks, add codspeed instead of pytest-benchmark, add action and gh workflow

* restore pytest-benchmark

* remove codspeed

* pin pydantic version, restore codspeed

* change on push to pydantic_v2 to trigger first one

* Use lifespan function instead of event (#1259)

* check return types

* fix imports order, set warnings=False on json that passes the dict, fix unnecessary loop in one of the test

* remove references to model's meta as it's now ormar config, rename related methods too

* filter out pydantic serializer warnings

* remove choices leftovers

* remove leftovers after property_fields, keep only enough to exclude them in initialization

* add migration guide

* fix meta references

* downgrade databases for now

* Change line numbers in documentation (#1265)

* proofread and fix the docs, part 1

* proofread and fix the docs for models

* proofread and fix the docs for fields

* proofread and fix the docs for relations

* proofread and fix rest of the docs, add release notes for 0.20

* create tables in new docs src

* cleanup old deps, uncomment docs publish on tag

* fix import reorder

---------

Co-authored-by: TouwaStar <30479449+TouwaStar@users.noreply.github.com>
Co-authored-by: Goran Mekić <meka@tilda.center>
This commit is contained in:
collerek
2024-03-23 19:28:28 +01:00
committed by GitHub
parent 3a206dd8dc
commit 500625f0ec
294 changed files with 8132 additions and 9311 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
tests/test_fastapi/test_binary_fields.py
View File

@ -1,63 +1,44 @@
import base64
import json
import uuid
from enum import Enum
from typing import List
import databases
import ormar
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
import ormar
from tests.settings import DATABASE_URL
app = FastAPI()
database = databases.Database(DATABASE_URL, force_rollback=True)
metadata = sqlalchemy.MetaData()
app.state.database = database
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
headers = {"content-type": "application/json"}
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
blob3 = b"\xc3\x28"
blob3 = b"\xc3\x83\x28"
blob4 = b"\xf0\x28\x8c\x28"
blob5 = b"\xee"
blob6 = b"\xff"
class BaseMeta(ormar.ModelMeta):
metadata = metadata
database = database
class BinaryEnum(Enum):
blob3 = blob3
blob4 = blob4
blob5 = blob5
blob6 = blob6
class BinaryThing(ormar.Model):
class Meta(BaseMeta):
tablename = "things"
ormar_config = base_ormar_config.copy(tablename="things")
id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4)
name: str = ormar.Text(default="")
bt: str = ormar.LargeBinary(
max_length=1000,
choices=[blob3, blob4, blob5, blob6],
represent_as_base64_str=True,
)
bt: str = ormar.LargeBinary(represent_as_base64_str=True, max_length=100)
create_test_database = init_tests(base_ormar_config)
@app.get("/things", response_model=List[BinaryThing])
@ -71,14 +52,6 @@ async def create_things(thing: BinaryThing):
return thing
@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.mark.asyncio
async def test_read_main():
client = AsyncClient(app=app, base_url="http://testserver")
@ -90,17 +63,15 @@ async def test_read_main():
)
assert response.status_code == 200
response = await client.get("/things")
assert response.json()[0]["bt"] == base64.b64encode(blob3).decode()
thing = BinaryThing(**response.json()[0])
assert response.json()[0]["bt"] == blob3.decode()
resp_json = response.json()
resp_json[0]["bt"] = resp_json[0]["bt"].encode()
thing = BinaryThing(**resp_json[0])
assert thing.__dict__["bt"] == blob3
assert thing.bt == base64.b64encode(blob3).decode()
def test_schema():
schema = BinaryThing.schema()
schema = BinaryThing.model_json_schema()
assert schema["properties"]["bt"]["format"] == "base64"
converted_choices = ["7g==", "/w==", "8CiMKA==", "wyg="]
assert len(schema["properties"]["bt"]["enum"]) == 4
assert all(
choice in schema["properties"]["bt"]["enum"] for choice in converted_choices
)
assert schema["example"]["bt"] == "string"

151
tests/test_fastapi/test_choices_schema.py
View File

@ -1,151 +0,0 @@
import datetime
import decimal
import uuid
from enum import Enum
import databases
import pydantic
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
import ormar
from tests.settings import DATABASE_URL
app = FastAPI()
database = databases.Database(DATABASE_URL, force_rollback=True)
metadata = sqlalchemy.MetaData()
app.state.database = database
uuid1 = uuid.uuid4()
uuid2 = uuid.uuid4()
blob = b"test"
blob2 = b"test2icac89uc98"
class EnumTest(Enum):
val1 = "Val1"
val2 = "Val2"
class Organisation(ormar.Model):
class Meta:
tablename = "org"
metadata = metadata
database = database
id: int = ormar.Integer(primary_key=True)
ident: str = ormar.String(max_length=100, choices=["ACME Ltd", "Other ltd"])
priority: int = ormar.Integer(choices=[1, 2, 3, 4, 5])
priority2: int = ormar.BigInteger(choices=[1, 2, 3, 4, 5])
priority3: int = ormar.SmallInteger(choices=[1, 2, 3, 4, 5])
expire_date: datetime.date = ormar.Date(
choices=[datetime.date(2021, 1, 1), datetime.date(2022, 5, 1)]
)
expire_time: datetime.time = ormar.Time(
choices=[datetime.time(10, 0, 0), datetime.time(12, 30)]
)
expire_datetime: datetime.datetime = ormar.DateTime(
choices=[
datetime.datetime(2021, 1, 1, 10, 0, 0),
datetime.datetime(2022, 5, 1, 12, 30),
]
)
random_val: float = ormar.Float(choices=[2.0, 3.5])
random_decimal: decimal.Decimal = ormar.Decimal(
scale=2, precision=4, choices=[decimal.Decimal(12.4), decimal.Decimal(58.2)]
)
random_json: pydantic.Json = ormar.JSON(choices=["aa", '{"aa": "bb"}'])
random_uuid: uuid.UUID = ormar.UUID(choices=[uuid1, uuid2])
enum_string: str = ormar.String(max_length=100, choices=list(EnumTest))
blob_col: bytes = ormar.LargeBinary(max_length=100000, choices=[blob, blob2])
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
@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)
@app.post("/items/", response_model=Organisation)
async def create_item(item: Organisation):
await item.save()
return item
@pytest.mark.asyncio
async def test_all_endpoints():
client = AsyncClient(app=app, base_url="http://testserver")
async with client as client, LifespanManager(app):
response = await client.post(
"/items/",
json={"id": 1, "ident": "", "priority": 4, "expire_date": "2022-05-01"},
)
assert response.status_code == 422
response = await client.post(
"/items/",
json={
"id": 1,
"ident": "ACME Ltd",
"priority": 4,
"priority2": 2,
"priority3": 1,
"expire_date": "2022-05-01",
"expire_time": "10:00:00",
"expire_datetime": "2022-05-01T12:30:00",
"random_val": 3.5,
"random_decimal": 12.4,
"random_json": '{"aa": "bb"}',
"random_uuid": str(uuid1),
"enum_string": EnumTest.val1.value,
"blob_col": blob.decode("utf-8"),
},
)
assert response.status_code == 200
item = Organisation(**response.json())
assert item.pk is not None
response = await client.get("/docs")
assert response.status_code == 200
assert b"<title>FastAPI - Swagger UI</title>" in response.content
def test_schema_modification():
schema = Organisation.schema()
for field in ["ident", "priority", "expire_date"]:
assert field in schema["properties"]
assert schema["properties"].get(field).get("enum") == list(
Organisation.Meta.model_fields.get(field).choices
)
assert "An enumeration." in schema["properties"].get(field).get("description")
def test_schema_gen():
schema = app.openapi()
assert "Organisation" in schema["components"]["schemas"]
props = schema["components"]["schemas"]["Organisation"]["properties"]
for field in [k for k in Organisation.Meta.model_fields.keys() if k != "id"]:
assert "enum" in props.get(field)
assert "description" in props.get(field)
assert "An enumeration." in props.get(field).get("description")

36
tests/test_fastapi/test_docs_with_multiple_relations_to_one.py
View File

@ -1,37 +1,28 @@
from typing import Optional
from uuid import UUID, uuid4
import databases
import ormar
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
import ormar
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
app = FastAPI()
DATABASE_URL = "sqlite:///db.sqlite"
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()
class BaseMeta(ormar.ModelMeta):
metadata = metadata
database = database
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
class CA(ormar.Model):
class Meta(BaseMeta):
tablename = "cas"
ormar_config = base_ormar_config.copy(tablename="cas")
id: UUID = ormar.UUID(primary_key=True, default=uuid4)
ca_name: str = ormar.Text(default="")
class CB1(ormar.Model):
class Meta(BaseMeta):
tablename = "cb1s"
ormar_config = base_ormar_config.copy(tablename="cb1s")
id: UUID = ormar.UUID(primary_key=True, default=uuid4)
cb1_name: str = ormar.Text(default="")
@ -39,14 +30,16 @@ class CB1(ormar.Model):
class CB2(ormar.Model):
class Meta(BaseMeta):
tablename = "cb2s"
ormar_config = base_ormar_config.copy(tablename="cb2s")
id: UUID = ormar.UUID(primary_key=True, default=uuid4)
cb2_name: str = ormar.Text(default="")
ca2: Optional[CA] = ormar.ForeignKey(CA, nullable=True)
create_test_database = init_tests(base_ormar_config)
@app.get("/ca", response_model=CA)
async def get_ca(): # pragma: no cover
return None
@ -70,6 +63,7 @@ async def test_all_endpoints():
assert response.status_code == 200, response.text
schema = response.json()
components = schema["components"]["schemas"]
assert all(x in components for x in ["CA", "CB1", "CB2"])
pk_onlys = [x for x in list(components.keys()) if x.startswith("PkOnly")]
assert len(pk_onlys) == 2
raw_names_w_o_modules = [x.split("__")[-1] for x in components.keys()]
assert all(x in raw_names_w_o_modules for x in ["CA", "CB1", "CB2"])
pk_onlys = [x for x in list(raw_names_w_o_modules) if x.startswith("PkOnly")]
assert len(pk_onlys) == 4

27
tests/test_fastapi/test_enum_schema.py
View File

@ -1,13 +1,11 @@
from enum import Enum
import databases
import sqlalchemy
import ormar
from tests.settings import DATABASE_URL
database = databases.Database(DATABASE_URL, force_rollback=True)
metadata = sqlalchemy.MetaData()
from tests.lifespan import init_tests
from tests.settings import create_config
base_ormar_config = create_config()
class MyEnum(Enum):
@ -16,18 +14,17 @@ class MyEnum(Enum):
class EnumExample(ormar.Model):
class Meta:
tablename = "enum_example"
metadata = metadata
database = database
ormar_config = base_ormar_config.copy(tablename="enum_example")
id: int = ormar.Integer(primary_key=True)
size: MyEnum = ormar.Enum(enum_class=MyEnum, default=MyEnum.SMALL)
create_test_database = init_tests(base_ormar_config)
def test_proper_schema():
schema = EnumExample.schema_json()
assert (
'{"MyEnum": {"title": "MyEnum", "description": "An enumeration.", '
'"enum": [1, 2]}}' in schema
)
schema = EnumExample.model_json_schema()
assert {"MyEnum": {"title": "MyEnum", "enum": [1, 2], "type": "integer"}} == schema[
"$defs"
]

76
tests/test_fastapi/test_excludes_with_get_pydantic.py
View File

@ -1,45 +1,49 @@
from typing import ForwardRef, Optional
import ormar
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
from tests.settings import DATABASE_URL
from tests.test_inheritance_and_pydantic_generation.test_geting_pydantic_models import (
Category,
SelfRef,
database,
metadata,
) # type: ignore
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
app = FastAPI()
app.state.database = database
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
class SelfRef(ormar.Model):
ormar_config = base_ormar_config.copy(tablename="self_refs")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100, default="selfref")
parent = ormar.ForeignKey(ForwardRef("SelfRef"), related_name="children")
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
SelfRef.update_forward_refs()
@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)
class Category(ormar.Model):
ormar_config = base_ormar_config.copy(tablename="categories")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
class Item(ormar.Model):
ormar_config = base_ormar_config.copy()
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100, default="test")
category: Optional[Category] = ormar.ForeignKey(Category, nullable=True)
create_test_database = init_tests(base_ormar_config)
async def create_category(category: Category):
return await Category(**category.dict()).save()
return await Category(**category.model_dump()).save()
create_category.__annotations__["category"] = Category.get_pydantic(exclude={"id"})
@ -55,7 +59,7 @@ async def create_selfref(
exclude={"children__name"} # noqa: F821
),
):
selfr = SelfRef(**selfref.dict())
selfr = SelfRef(**selfref.model_dump())
await selfr.save()
if selfr.children:
for child in selfr.children:
@ -107,12 +111,12 @@ async def test_read_main():
assert self_ref.id == 3
assert self_ref.name == "test3"
assert self_ref.parent is None
assert self_ref.children[0].dict() == {"id": 4}
assert self_ref.children[0].model_dump() == {"id": 4}
response = await client.get("/selfrefs/3/")
assert response.status_code == 200
check_children = SelfRef(**response.json())
assert check_children.children[0].dict() == {
assert check_children.children[0].model_dump() == {
"children": [],
"id": 4,
"name": "selfref",
@ -122,9 +126,19 @@ async def test_read_main():
response = await client.get("/selfrefs/2/")
assert response.status_code == 200
check_children = SelfRef(**response.json())
assert check_children.dict() == {
assert check_children.model_dump() == {
"children": [],
"id": 2,
"name": "test2",
"parent": {"id": 1},
}
response = await client.get("/selfrefs/1/")
assert response.status_code == 200
check_children = SelfRef(**response.json())
assert check_children.model_dump() == {
"children": [{"id": 2, "name": "test2"}],
"id": 1,
"name": "test",
"parent": None,
}

71
tests/test_fastapi/test_excluding_fields.py
View File

@ -1,62 +1,34 @@
from typing import List
import databases
import ormar
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
import ormar
from tests.settings import DATABASE_URL
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
app = FastAPI()
metadata = sqlalchemy.MetaData()
database = databases.Database(DATABASE_URL, force_rollback=True)
app.state.database = database
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
class Category(ormar.Model):
class Meta:
tablename = "categories"
metadata = metadata
database = database
ormar_config = base_ormar_config.copy(tablename="categories")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
class Item(ormar.Model):
class Meta:
tablename = "items"
metadata = metadata
database = database
ormar_config = base_ormar_config.copy(tablename="items")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
categories: List[Category] = ormar.ManyToMany(Category)
@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)
create_test_database = init_tests(base_ormar_config)
@app.post("/items/", response_model=Item)
@ -68,40 +40,26 @@ async def create_item(item: Item):
@app.get("/items/{item_id}")
async def get_item(item_id: int):
item = await Item.objects.select_related("categories").get(pk=item_id)
return item.dict(exclude_primary_keys=True, exclude_through_models=True)
return item.model_dump(exclude_primary_keys=True, exclude_through_models=True)
@app.get("/categories/{category_id}")
async def get_category(category_id: int):
category = await Category.objects.select_related("items").get(pk=category_id)
return category.dict(exclude_primary_keys=True)
return category.model_dump(exclude_primary_keys=True)
@app.get("/categories/nt/{category_id}")
async def get_category_no_through(category_id: int):
category = await Category.objects.select_related("items").get(pk=category_id)
return category.dict(exclude_through_models=True)
result = category.model_dump(exclude_through_models=True)
return result
@app.get("/categories/ntp/{category_id}")
async def get_category_no_pk_through(category_id: int):
category = await Category.objects.select_related("items").get(pk=category_id)
return category.dict(exclude_through_models=True, exclude_primary_keys=True)
@app.get(
"/items/fex/{item_id}",
response_model=Item,
response_model_exclude={
"id",
"categories__id",
"categories__itemcategory",
"categories__items",
},
)
async def get_item_excl(item_id: int):
item = await Item.objects.select_all().get(pk=item_id)
return item
return category.model_dump(exclude_through_models=True, exclude_primary_keys=True)
@pytest.mark.asyncio
@ -120,9 +78,6 @@ async def test_all_endpoints():
no_pk_item = (await client.get(f"/items/{item_check.id}")).json()
assert no_pk_item == item
no_pk_item2 = (await client.get(f"/items/fex/{item_check.id}")).json()
assert no_pk_item2 == item
no_pk_category = (
await client.get(f"/categories/{item_check.categories[0].id}")
).json()

41
tests/test_fastapi/test_extra_ignore_parameter.py
View File

@ -1,52 +1,25 @@
import json
import databases
import ormar
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
import ormar
from ormar import Extra
from tests.settings import DATABASE_URL
app = FastAPI()
metadata = sqlalchemy.MetaData()
database = databases.Database(DATABASE_URL, force_rollback=True)
app.state.database = database
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
class Item(ormar.Model):
class Meta:
database = database
metadata = metadata
extra = Extra.ignore
ormar_config = base_ormar_config.copy(extra=Extra.ignore)
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
@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)
create_test_database = init_tests(base_ormar_config)
@app.post("/item/", response_model=Item)

82
tests/test_fastapi/test_fastapi_docs.py
View File

@ -1,40 +1,19 @@
import datetime
from typing import List, Optional
from typing import List, Optional, Union
import databases
import ormar
import pydantic
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
from pydantic import Field
import ormar
from tests.settings import DATABASE_URL
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
app = FastAPI()
metadata = sqlalchemy.MetaData()
database = databases.Database(DATABASE_URL, force_rollback=True)
app.state.database = database
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
class LocalMeta:
metadata = metadata
database = database
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
class PTestA(pydantic.BaseModel):
@ -49,35 +28,31 @@ class PTestP(pydantic.BaseModel):
class Category(ormar.Model):
class Meta(LocalMeta):
tablename = "categories"
ormar_config = base_ormar_config.copy(tablename="categories")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
class Item(ormar.Model):
class Meta(LocalMeta):
pass
ormar_config = base_ormar_config.copy()
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
pydantic_int: Optional[int]
test_P: Optional[List[PTestP]]
pydantic_int: Optional[int] = None
test_P: List[PTestP] = Field(default_factory=list)
test_P_or_A: Union[int, str, None] = None
categories = ormar.ManyToMany(Category)
@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)
create_test_database = init_tests(base_ormar_config)
@app.get("/items/", response_model=List[Item])
async def get_items():
items = await Item.objects.select_related("categories").all()
for item in items:
item.test_P_or_A = 2
return items
@ -104,26 +79,35 @@ async def test_all_endpoints():
client = AsyncClient(app=app, base_url="http://testserver")
async with client as client, LifespanManager(app):
response = await client.post("/categories/", json={"name": "test cat"})
assert response.status_code == 200
category = response.json()
response = await client.post("/categories/", json={"name": "test cat2"})
assert response.status_code == 200
category2 = response.json()
response = await client.post("/items/", json={"name": "test", "id": 1})
response = await client.post(
"/items/", json={"name": "test", "id": 1, "test_P_or_A": 0}
)
assert response.status_code == 200
item = Item(**response.json())
assert item.pk is not None
response = await client.post(
"/items/add_category/", json={"item": item.dict(), "category": category}
"/items/add_category/",
json={"item": item.model_dump(), "category": category},
)
assert response.status_code == 200
item = Item(**response.json())
assert len(item.categories) == 1
assert item.categories[0].name == "test cat"
await client.post(
"/items/add_category/", json={"item": item.dict(), "category": category2}
"/items/add_category/",
json={"item": item.model_dump(), "category": category2},
)
response = await client.get("/items/")
assert response.status_code == 200
items = [Item(**item) for item in response.json()]
assert items[0] == item
assert len(items[0].categories) == 2
@ -136,7 +120,7 @@ async def test_all_endpoints():
def test_schema_modification():
schema = Item.schema()
schema = Item.model_json_schema()
assert any(
x.get("type") == "array" for x in schema["properties"]["categories"]["anyOf"]
)
@ -147,10 +131,11 @@ def test_schema_modification():
"name": "string",
"pydantic_int": 0,
"test_P": [{"a": 0, "b": {"c": "string", "d": "string", "e": "string"}}],
"test_P_or_A": (0, "string"),
}
schema = Category.schema()
assert schema["example"] == {
schema = Category.model_json_schema()
assert schema["$defs"]["Category"]["example"] == {
"id": 0,
"name": "string",
"items": [
@ -161,6 +146,7 @@ def test_schema_modification():
"test_P": [
{"a": 0, "b": {"c": "string", "d": "string", "e": "string"}}
],
"test_P_or_A": (0, "string"),
}
],
}
@ -169,4 +155,6 @@ def test_schema_modification():
def test_schema_gen():
schema = app.openapi()
assert "Category" in schema["components"]["schemas"]
assert "Item" in schema["components"]["schemas"]
subschemas = [x.split("__")[-1] for x in schema["components"]["schemas"]]
assert "Item-Input" in subschemas
assert "Item-Output" in subschemas

28
tests/test_fastapi/test_fastapi_usage.py
View File

@ -1,42 +1,36 @@
from typing import Optional
import databases
import ormar
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
import ormar
from tests.settings import DATABASE_URL
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
app = FastAPI()
database = databases.Database(DATABASE_URL, force_rollback=True)
metadata = sqlalchemy.MetaData()
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
class Category(ormar.Model):
class Meta:
tablename = "categories"
metadata = metadata
database = database
ormar_config = base_ormar_config.copy(tablename="categories")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
class Item(ormar.Model):
class Meta:
tablename = "items"
metadata = metadata
database = database
ormar_config = base_ormar_config.copy(tablename="items")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
category: Optional[Category] = ormar.ForeignKey(Category, nullable=True)
create_test_database = init_tests(base_ormar_config)
@app.post("/items/", response_model=Item)
async def create_item(item: Item):
return item
@ -53,7 +47,6 @@ async def test_read_main():
assert response.json() == {
"category": {
"id": None,
"items": [{"id": 1, "name": "test"}],
"name": "test cat",
},
"id": 1,
@ -61,3 +54,4 @@ async def test_read_main():
}
item = Item(**response.json())
assert item.id == 1
assert item.category.items[0].id == 1

187
tests/test_fastapi/test_inheritance_concrete_fastapi.py
View File

@ -1,41 +1,139 @@
import datetime
from typing import List
from typing import List, Optional
import ormar
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
from ormar.relations.relation_proxy import RelationProxy
from pydantic import computed_field
from tests.settings import DATABASE_URL
from tests.test_inheritance_and_pydantic_generation.test_inheritance_concrete import ( # type: ignore
Category,
Subject,
Person,
Bus,
Truck,
Bus2,
Truck2,
db as database,
metadata,
)
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
app = FastAPI()
app.state.database = database
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
class AuditModel(ormar.Model):
ormar_config = base_ormar_config.copy(abstract=True)
created_by: str = ormar.String(max_length=100)
updated_by: str = ormar.String(max_length=100, default="Sam")
@computed_field
def audit(self) -> str: # pragma: no cover
return f"{self.created_by} {self.updated_by}"
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
class DateFieldsModelNoSubclass(ormar.Model):
ormar_config = base_ormar_config.copy(tablename="test_date_models")
date_id: int = ormar.Integer(primary_key=True)
created_date: datetime.datetime = ormar.DateTime(default=datetime.datetime.now)
updated_date: datetime.datetime = ormar.DateTime(default=datetime.datetime.now)
class DateFieldsModel(ormar.Model):
ormar_config = base_ormar_config.copy(
abstract=True,
constraints=[
ormar.fields.constraints.UniqueColumns(
"creation_date",
"modification_date",
),
ormar.fields.constraints.CheckColumns(
"creation_date <= modification_date",
),
],
)
created_date: datetime.datetime = ormar.DateTime(
default=datetime.datetime.now, name="creation_date"
)
updated_date: datetime.datetime = ormar.DateTime(
default=datetime.datetime.now, name="modification_date"
)
class Person(ormar.Model):
ormar_config = base_ormar_config.copy()
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
class Car(ormar.Model):
ormar_config = base_ormar_config.copy(abstract=True)
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=50)
owner: Person = ormar.ForeignKey(Person)
co_owner: Person = ormar.ForeignKey(Person, related_name="coowned")
created_date: datetime.datetime = ormar.DateTime(default=datetime.datetime.now)
class Car2(ormar.Model):
ormar_config = base_ormar_config.copy(abstract=True)
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=50)
owner: Person = ormar.ForeignKey(Person, related_name="owned")
co_owners: RelationProxy[Person] = ormar.ManyToMany(Person, related_name="coowned")
created_date: datetime.datetime = ormar.DateTime(default=datetime.datetime.now)
class Bus(Car):
ormar_config = base_ormar_config.copy(tablename="buses")
owner: Person = ormar.ForeignKey(Person, related_name="buses")
max_persons: int = ormar.Integer()
class Bus2(Car2):
ormar_config = base_ormar_config.copy(tablename="buses2")
max_persons: int = ormar.Integer()
class Category(DateFieldsModel, AuditModel):
ormar_config = base_ormar_config.copy(tablename="categories")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=50, unique=True, index=True)
code: int = ormar.Integer()
@computed_field
def code_name(self) -> str:
return f"{self.code}:{self.name}"
@computed_field
def audit(self) -> str:
return f"{self.created_by} {self.updated_by}"
class Subject(DateFieldsModel):
ormar_config = base_ormar_config.copy()
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=50, unique=True, index=True)
category: Optional[Category] = ormar.ForeignKey(Category)
class Truck(Car):
ormar_config = base_ormar_config.copy()
max_capacity: int = ormar.Integer()
class Truck2(Car2):
ormar_config = base_ormar_config.copy(tablename="trucks2")
max_capacity: int = ormar.Integer()
create_test_database = init_tests(base_ormar_config)
@app.post("/subjects/", response_model=Subject)
@ -111,14 +209,6 @@ async def add_truck_coowner(item_id: int, person: Person):
return truck
@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.mark.asyncio
async def test_read_main():
client = AsyncClient(app=app, base_url="http://testserver")
@ -134,7 +224,7 @@ async def test_read_main():
assert cat.created_date is not None
assert cat.id == 1
cat_dict = cat.dict()
cat_dict = cat.model_dump()
cat_dict["updated_date"] = cat_dict["updated_date"].strftime(
"%Y-%m-%d %H:%M:%S.%f"
)
@ -160,11 +250,14 @@ async def test_inheritance_with_relation():
truck_dict = dict(
name="Shelby wanna be",
max_capacity=1400,
owner=sam.dict(),
co_owner=joe.dict(),
owner=sam.model_dump(),
co_owner=joe.model_dump(),
)
bus_dict = dict(
name="Unicorn", max_persons=50, owner=sam.dict(), co_owner=joe.dict()
name="Unicorn",
max_persons=50,
owner=sam.model_dump(),
co_owner=joe.model_dump(),
)
unicorn = Bus(**(await client.post("/buses/", json=bus_dict)).json())
shelby = Truck(**(await client.post("/trucks/", json=truck_dict)).json())
@ -201,21 +294,25 @@ async def test_inheritance_with_m2m_relation():
joe = Person(**(await client.post("/persons/", json={"name": "Joe"})).json())
alex = Person(**(await client.post("/persons/", json={"name": "Alex"})).json())
truck_dict = dict(name="Shelby wanna be", max_capacity=2000, owner=sam.dict())
bus_dict = dict(name="Unicorn", max_persons=80, owner=sam.dict())
truck_dict = dict(
name="Shelby wanna be", max_capacity=2000, owner=sam.model_dump()
)
bus_dict = dict(name="Unicorn", max_persons=80, owner=sam.model_dump())
unicorn = Bus2(**(await client.post("/buses2/", json=bus_dict)).json())
shelby = Truck2(**(await client.post("/trucks2/", json=truck_dict)).json())
unicorn = Bus2(
**(
await client.post(f"/buses2/{unicorn.pk}/add_coowner/", json=joe.dict())
await client.post(
f"/buses2/{unicorn.pk}/add_coowner/", json=joe.model_dump()
)
).json()
)
unicorn = Bus2(
**(
await client.post(
f"/buses2/{unicorn.pk}/add_coowner/", json=alex.dict()
f"/buses2/{unicorn.pk}/add_coowner/", json=alex.model_dump()
)
).json()
)
@ -232,11 +329,13 @@ async def test_inheritance_with_m2m_relation():
assert unicorn.co_owners[1] == alex
assert unicorn.max_persons == 80
await client.post(f"/trucks2/{shelby.pk}/add_coowner/", json=alex.dict())
await client.post(f"/trucks2/{shelby.pk}/add_coowner/", json=alex.model_dump())
shelby = Truck2(
**(
await client.post(f"/trucks2/{shelby.pk}/add_coowner/", json=joe.dict())
await client.post(
f"/trucks2/{shelby.pk}/add_coowner/", json=joe.model_dump()
)
).json()
)

56
tests/test_fastapi/test_inheritance_mixins_fastapi.py
View File

@ -1,30 +1,46 @@
import datetime
from typing import Optional
import ormar
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
from tests.settings import DATABASE_URL
from tests.test_inheritance_and_pydantic_generation.test_inheritance_mixins import Category, Subject, metadata, db as database # type: ignore
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
app = FastAPI()
app.state.database = database
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
class AuditMixin:
created_by: str = ormar.String(max_length=100)
updated_by: str = ormar.String(max_length=100, default="Sam")
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
class DateFieldsMixins:
created_date: datetime.datetime = ormar.DateTime(default=datetime.datetime.now)
updated_date: datetime.datetime = ormar.DateTime(default=datetime.datetime.now)
class Category(ormar.Model, DateFieldsMixins, AuditMixin):
ormar_config = base_ormar_config.copy(tablename="categories")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=50, unique=True, index=True)
code: int = ormar.Integer()
class Subject(ormar.Model, DateFieldsMixins):
ormar_config = base_ormar_config.copy(tablename="subjects")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=50, unique=True, index=True)
category: Optional[Category] = ormar.ForeignKey(Category)
create_test_database = init_tests(base_ormar_config)
@app.post("/subjects/", response_model=Subject)
@ -38,14 +54,6 @@ async def create_category(category: Category):
return category
@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.mark.asyncio
async def test_read_main():
client = AsyncClient(app=app, base_url="http://testserver")
@ -61,7 +69,7 @@ async def test_read_main():
assert cat.created_date is not None
assert cat.id == 1
cat_dict = cat.dict()
cat_dict = cat.model_dump()
cat_dict["updated_date"] = cat_dict["updated_date"].strftime(
"%Y-%m-%d %H:%M:%S.%f"
)

62
tests/test_fastapi/test_json_field_fastapi.py
View File

@ -2,52 +2,31 @@
import uuid
from typing import List
import databases
import ormar
import pydantic
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
import ormar
from tests.settings import DATABASE_URL
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
app = FastAPI()
database = databases.Database(DATABASE_URL, force_rollback=True)
metadata = sqlalchemy.MetaData()
app.state.database = database
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
class BaseMeta(ormar.ModelMeta):
metadata = metadata
database = database
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
class Thing(ormar.Model):
class Meta(BaseMeta):
tablename = "things"
ormar_config = base_ormar_config.copy(tablename="things")
id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4)
name: str = ormar.Text(default="")
js: pydantic.Json = ormar.JSON()
create_test_database = init_tests(base_ormar_config)
@app.get("/things", response_model=List[Thing])
async def read_things():
return await Thing.objects.order_by("name").all()
@ -87,14 +66,6 @@ async def read_things_untyped():
return await Thing.objects.order_by("name").all()
@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.mark.asyncio
async def test_json_is_required_if_not_nullable():
with pytest.raises(pydantic.ValidationError):
@ -104,8 +75,7 @@ async def test_json_is_required_if_not_nullable():
@pytest.mark.asyncio
async def test_json_is_not_required_if_nullable():
class Thing2(ormar.Model):
class Meta(BaseMeta):
tablename = "things2"
ormar_config = base_ormar_config.copy(tablename="things2")
id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4)
name: str = ormar.Text(default="")
@ -116,16 +86,16 @@ async def test_json_is_not_required_if_nullable():
@pytest.mark.asyncio
async def test_setting_values_after_init():
async with database:
async with base_ormar_config.database:
t1 = Thing(id="67a82813-d90c-45ff-b546-b4e38d7030d7", name="t1", js=["thing1"])
assert '["thing1"]' in t1.json()
assert '["thing1"]' in t1.model_dump_json()
await t1.save()
t1.json()
assert '["thing1"]' in t1.json()
t1.model_dump_json()
assert '["thing1"]' in t1.model_dump_json()
assert '["thing1"]' in (await Thing.objects.get(id=t1.id)).json()
assert '["thing1"]' in (await Thing.objects.get(id=t1.id)).model_dump_json()
await t1.update()
assert '["thing1"]' in (await Thing.objects.get(id=t1.id)).json()
assert '["thing1"]' in (await Thing.objects.get(id=t1.id)).model_dump_json()
@pytest.mark.asyncio

58
tests/test_fastapi/test_m2m_forwardref.py
View File

@ -1,42 +1,17 @@
from typing import List, Optional
import databases
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from pydantic.schema import ForwardRef
from starlette import status
from httpx import AsyncClient
from typing import ForwardRef, List, Optional
import ormar
import pytest
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
from starlette import status
app = FastAPI()
from tests.settings import DATABASE_URL
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
database = databases.Database(DATABASE_URL, force_rollback=True)
metadata = sqlalchemy.MetaData()
app.state.database = database
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
class BaseMeta(ormar.ModelMeta):
database = database
metadata = metadata
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
CityRef = ForwardRef("City")
@ -45,8 +20,7 @@ CountryRef = ForwardRef("Country")
# models.py
class Country(ormar.Model):
class Meta(BaseMeta):
tablename = "countries"
ormar_config = base_ormar_config.copy(tablename="countries")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=128, unique=True)
@ -64,8 +38,7 @@ class Country(ormar.Model):
class City(ormar.Model):
class Meta(BaseMeta):
tablename = "cities"
ormar_config = base_ormar_config.copy(tablename="cities")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=128)
@ -77,12 +50,7 @@ class City(ormar.Model):
Country.update_forward_refs()
@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)
create_test_database = init_tests(base_ormar_config)
@app.post("/", response_model=Country, status_code=status.HTTP_201_CREATED)

50
tests/test_fastapi/test_more_reallife_fastapi.py
View File

@ -1,62 +1,34 @@
from typing import List, Optional
import databases
import ormar
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
import ormar
from tests.settings import DATABASE_URL
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
app = FastAPI()
metadata = sqlalchemy.MetaData()
database = databases.Database(DATABASE_URL, force_rollback=True)
app.state.database = database
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
class Category(ormar.Model):
class Meta:
tablename = "categories"
metadata = metadata
database = database
ormar_config = base_ormar_config.copy(tablename="categories")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
class Item(ormar.Model):
class Meta:
tablename = "items"
metadata = metadata
database = database
ormar_config = base_ormar_config.copy(tablename="items")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
category: Optional[Category] = ormar.ForeignKey(Category, nullable=True)
@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)
create_test_database = init_tests(base_ormar_config)
@app.get("/items", response_model=List[Item])
@ -92,7 +64,7 @@ async def get_item(item_id: int):
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
item_db = await Item.objects.get(pk=item_id)
return await item_db.update(**item.dict())
return await item_db.update(**item.model_dump())
@app.delete("/items/{item_id}")
@ -118,8 +90,8 @@ async def test_all_endpoints():
assert items[0] == item
item.name = "New name"
response = await client.put(f"/items/{item.pk}", json=item.dict())
assert response.json() == item.dict()
response = await client.put(f"/items/{item.pk}", json=item.model_dump())
assert response.json() == item.model_dump()
response = await client.get("/items")
items = [Item(**item) for item in response.json()]

59
tests/test_fastapi/test_nested_saving.py
View File

@ -1,52 +1,29 @@
import json
from typing import Any, Dict, Optional, Set, Type, Union, cast
import databases
import ormar
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
import ormar
from ormar.queryset.utils import translate_list_to_dict
from tests.settings import DATABASE_URL
app = FastAPI()
metadata = sqlalchemy.MetaData()
database = databases.Database(DATABASE_URL, force_rollback=True)
app.state.database = database
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
headers = {"content-type": "application/json"}
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
class Department(ormar.Model):
class Meta:
database = database
metadata = metadata
ormar_config = base_ormar_config.copy()
id: int = ormar.Integer(primary_key=True)
department_name: str = ormar.String(max_length=100)
class Course(ormar.Model):
class Meta:
database = database
metadata = metadata
ormar_config = base_ormar_config.copy()
id: int = ormar.Integer(primary_key=True)
course_name: str = ormar.String(max_length=100)
@ -55,24 +32,13 @@ class Course(ormar.Model):
class Student(ormar.Model):
class Meta:
database = database
metadata = metadata
ormar_config = base_ormar_config.copy()
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
courses = ormar.ManyToMany(Course)
# create db and tables
@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)
to_exclude = {
"id": ...,
"courses": {
@ -88,6 +54,9 @@ to_exclude_ormar = {
}
create_test_database = init_tests(base_ormar_config)
def auto_exclude_id_field(to_exclude: Any) -> Union[Dict, Set]:
if isinstance(to_exclude, dict):
for key in to_exclude.keys():
@ -117,7 +86,7 @@ async def get_department(department_name: str):
department = await Department.objects.select_all(follow=True).get(
department_name=department_name
)
return department.dict(exclude=to_exclude)
return department.model_dump(exclude=to_exclude)
@app.get("/departments/{department_name}/second")
@ -125,7 +94,7 @@ async def get_department_exclude(department_name: str):
department = await Department.objects.select_all(follow=True).get(
department_name=department_name
)
return department.dict(exclude=to_exclude_ormar)
return department.model_dump(exclude=to_exclude_ormar)
@app.get("/departments/{department_name}/exclude")
@ -133,7 +102,7 @@ async def get_department_exclude_all(department_name: str):
department = await Department.objects.select_all(follow=True).get(
department_name=department_name
)
return department.dict(exclude=exclude_all)
return department.model_dump(exclude=exclude_all)
@pytest.mark.asyncio

53
tests/test_fastapi/test_recursion_error.py
View File

@ -1,41 +1,22 @@
import json
import uuid
from datetime import datetime
from typing import List
import databases
import ormar
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import Depends, FastAPI
from httpx import AsyncClient
from pydantic import BaseModel, Json
import ormar
from tests.settings import DATABASE_URL
router = FastAPI()
metadata = sqlalchemy.MetaData()
database = databases.Database(DATABASE_URL, force_rollback=True)
router.state.database = database
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
base_ormar_config = create_config()
router = FastAPI(lifespan=lifespan(base_ormar_config))
headers = {"content-type": "application/json"}
@router.on_event("startup")
async def startup() -> None:
database_ = router.state.database
if not database_.is_connected:
await database_.connect()
@router.on_event("shutdown")
async def shutdown() -> None:
database_ = router.state.database
if database_.is_connected:
await database_.disconnect()
class User(ormar.Model):
"""
The user model
@ -49,10 +30,7 @@ class User(ormar.Model):
verify_key: str = ormar.String(unique=True, max_length=100, nullable=True)
created_at: datetime = ormar.DateTime(default=datetime.now())
class Meta:
tablename = "users"
metadata = metadata
database = database
ormar_config = base_ormar_config.copy(tablename="users")
class UserSession(ormar.Model):
@ -65,10 +43,7 @@ class UserSession(ormar.Model):
session_key: str = ormar.String(unique=True, max_length=64)
created_at: datetime = ormar.DateTime(default=datetime.now())
class Meta:
tablename = "user_sessions"
metadata = metadata
database = database
ormar_config = base_ormar_config.copy(tablename="user_sessions")
class QuizAnswer(BaseModel):
@ -96,18 +71,10 @@ class Quiz(ormar.Model):
user_id: uuid.UUID = ormar.UUID(foreign_key=User.id)
questions: Json = ormar.JSON(nullable=False)
class Meta:
tablename = "quiz"
metadata = metadata
database = database
ormar_config = base_ormar_config.copy(tablename="quiz")
@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)
create_test_database = init_tests(base_ormar_config)
async def get_current_user():
@ -118,7 +85,7 @@ async def get_current_user():
async def create_quiz_lol(
quiz_input: QuizInput, user: User = Depends(get_current_user)
):
quiz = Quiz(**quiz_input.dict(), user_id=user.id)
quiz = Quiz(**quiz_input.model_dump(), user_id=user.id)
return await quiz.save()

72
tests/test_fastapi/test_relations_with_nested_defaults.py
View File

@ -1,53 +1,28 @@
from typing import Optional
import databases
import ormar
import pytest
import pytest_asyncio
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
import ormar
from tests.settings import DATABASE_URL
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()
app = FastAPI()
app.state.database = database
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
class BaseMeta(ormar.ModelMeta):
metadata = metadata
database = database
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
class Country(ormar.Model):
class Meta(BaseMeta):
tablename = "countries"
ormar_config = base_ormar_config.copy(tablename="countries")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100, default="Poland")
class Author(ormar.Model):
class Meta(BaseMeta):
tablename = "authors"
ormar_config = base_ormar_config.copy(tablename="authors")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
@ -56,8 +31,7 @@ class Author(ormar.Model):
class Book(ormar.Model):
class Meta(BaseMeta):
tablename = "books"
ormar_config = base_ormar_config.copy(tablename="books")
id: int = ormar.Integer(primary_key=True)
author: Optional[Author] = ormar.ForeignKey(Author)
@ -65,17 +39,12 @@ class Book(ormar.Model):
year: int = ormar.Integer(nullable=True)
@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)
create_test_database = init_tests(base_ormar_config)
@pytest_asyncio.fixture
async def sample_data():
async with database:
async with base_ormar_config.database:
country = await Country(id=1, name="USA").save()
author = await Author(id=1, name="bug", rating=5, country=country).save()
await Book(
@ -101,7 +70,17 @@ async def test_related_with_defaults(sample_data):
async with client as client, LifespanManager(app):
response = await client.get("/books/1")
assert response.json() == {
"author": {"id": 1},
"author": {
"books": [
{
"author": {"id": 1},
"id": 1,
"title": "Bug caused by default value",
"year": 2021,
}
],
"id": 1,
},
"id": 1,
"title": "Bug caused by default value",
"year": 2021,
@ -111,9 +90,14 @@ async def test_related_with_defaults(sample_data):
assert response.json() == {
"author": {
"books": [
{"id": 1, "title": "Bug caused by default value", "year": 2021}
{
"author": {"id": 1},
"id": 1,
"title": "Bug caused by default value",
"year": 2021,
}
],
"country": {"id": 1},
"country": {"authors": [{"id": 1}], "id": 1},
"id": 1,
"name": "bug",
"rating": 5,

21
tests/test_fastapi/test_schema_not_allowed_params.py
View File

@ -1,29 +1,24 @@
import databases
import sqlalchemy
import ormar
DATABASE_URL = "sqlite:///db.sqlite"
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()
from tests.lifespan import init_tests
from tests.settings import create_config
class BaseMeta(ormar.ModelMeta):
metadata = metadata
database = database
base_ormar_config = create_config()
class Author(ormar.Model):
class Meta(BaseMeta):
tablename = "authors"
ormar_config = base_ormar_config.copy(tablename="authors")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
contents: str = ormar.Text()
create_test_database = init_tests(base_ormar_config)
def test_schema_not_allowed():
schema = Author.schema()
schema = Author.model_json_schema()
for field_schema in schema.get("properties").values():
for key in field_schema.keys():
assert "_" not in key, f"Found illegal field in openapi schema: {key}"

66
tests/test_fastapi/test_skip_reverse_models.py
View File

@ -1,45 +1,21 @@
from typing import List, Optional
import databases
import ormar
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
import ormar
from tests.settings import DATABASE_URL
app = FastAPI()
metadata = sqlalchemy.MetaData()
database = databases.Database(DATABASE_URL, force_rollback=True)
app.state.database = database
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
headers = {"content-type": "application/json"}
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
class BaseMeta(ormar.ModelMeta):
database = database
metadata = metadata
class Author(ormar.Model):
class Meta(BaseMeta):
pass
ormar_config = base_ormar_config.copy()
id: int = ormar.Integer(primary_key=True)
first_name: str = ormar.String(max_length=80)
@ -47,16 +23,18 @@ class Author(ormar.Model):
class Category(ormar.Model):
class Meta(BaseMeta):
tablename = "categories"
ormar_config = base_ormar_config.copy(tablename="categories")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=40)
class Category2(Category):
model_config = dict(extra="forbid")
class Post(ormar.Model):
class Meta(BaseMeta):
pass
ormar_config = base_ormar_config.copy()
id: int = ormar.Integer(primary_key=True)
title: str = ormar.String(max_length=200)
@ -64,12 +42,12 @@ class Post(ormar.Model):
author: Optional[Author] = ormar.ForeignKey(Author, skip_reverse=True)
@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)
create_test_database = init_tests(base_ormar_config)
@app.post("/categories/forbid/", response_model=Category2)
async def create_category_forbid(category: Category2): # pragma: no cover
pass
@app.post("/categories/", response_model=Category)
@ -115,7 +93,7 @@ async def test_queries():
assert response.status_code == 200
response = await client.get("/categories/")
assert response.status_code == 200
assert not "posts" in response.json()
assert "posts" not in response.json()
categories = [Category(**x) for x in response.json()]
assert categories[0] is not None
assert categories[0].name == "Test category2"
@ -139,7 +117,7 @@ async def test_queries():
response = await client.post("/posts/", json=right_post, headers=headers)
assert response.status_code == 200
Category.__config__.extra = "allow"
Category.model_config["extra"] = "allow"
response = await client.get("/posts/")
assert response.status_code == 200
posts = [Post(**x) for x in response.json()]
@ -150,6 +128,6 @@ async def test_queries():
wrong_category = {"name": "Test category3", "posts": [{"title": "Test Post"}]}
# cannot add posts if skipped, will be error with extra forbid
Category.__config__.extra = "forbid"
response = await client.post("/categories/", json=wrong_category)
assert Category2.model_config["extra"] == "forbid"
response = await client.post("/categories/forbid/", json=wrong_category)
assert response.status_code == 422

53
tests/test_fastapi/test_wekref_exclusion.py
View File

@ -1,55 +1,22 @@
from typing import List, Optional
from uuid import UUID, uuid4
import databases
import ormar
import pydantic
import pytest
import sqlalchemy
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
import ormar
from tests.settings import DATABASE_URL
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
app = FastAPI()
database = databases.Database(DATABASE_URL, force_rollback=True)
metadata = sqlalchemy.MetaData()
app.state.database = database
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
@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)
class BaseMeta(ormar.ModelMeta):
database = database
metadata = metadata
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
class OtherThing(ormar.Model):
class Meta(BaseMeta):
tablename = "other_things"
ormar_config = base_ormar_config.copy(tablename="other_things")
id: UUID = ormar.UUID(primary_key=True, default=uuid4)
name: str = ormar.Text(default="")
@ -57,8 +24,7 @@ class OtherThing(ormar.Model):
class Thing(ormar.Model):
class Meta(BaseMeta):
tablename = "things"
ormar_config = base_ormar_config.copy(tablename="things")
id: UUID = ormar.UUID(primary_key=True, default=uuid4)
name: str = ormar.Text(default="")
@ -66,6 +32,9 @@ class Thing(ormar.Model):
other_thing: Optional[OtherThing] = ormar.ForeignKey(OtherThing, nullable=True)
create_test_database = init_tests(base_ormar_config)
@app.post("/test/1")
async def post_test_1():
# don't split initialization and attribute assignment
@ -98,7 +67,7 @@ async def get_test_3():
ot = await OtherThing.objects.select_related("things").get()
# exclude unwanted field while ot is still in scope
# in order not to pass it to fastapi
return [t.dict(exclude={"other_thing"}) for t in ot.things]
return [t.model_dump(exclude={"other_thing"}) for t in ot.things]
@app.get("/test/4", response_model=List[Thing], response_model_exclude={"other_thing"})