Allow upserting rows with existing_pk even if they do not exist. (#889)

* Allow for __force_save__ in Model.upsert() method to save the models despite they already have a pk set. On integrity Error proceed to update the model, so in worst case two db calls will be made.

* Fix coverage

* Change implementation to checking if the row exists as postgres leaves hanging invalid transaction on integrity error. On force_save always check if row exists and then save/update (so always two queries).
This commit is contained in:
collerek
2022-10-21 13:32:36 +02:00
committed by GitHub
parent f94e507d70
commit 4fa0205598
4 changed files with 157 additions and 1 deletions

View File

@ -0,0 +1,83 @@
import uuid
from typing import 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 Department(ormar.Model):
class Meta:
database = database
metadata = metadata
id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4)
department_name: str = ormar.String(max_length=100)
class Course(ormar.Model):
class Meta:
database = database
metadata = metadata
id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4)
course_name: str = ormar.String(max_length=100)
completed: bool = ormar.Boolean()
department: Optional[Department] = ormar.ForeignKey(Department)
class Student(ormar.Model):
class Meta:
database = database
metadata = metadata
id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4)
name: str = ormar.String(max_length=100)
courses = ormar.ManyToMany(Course)
@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_uuid_pk_in_save_related():
async with database:
to_save = {
"department_name": "Ormar",
"courses": [
{
"course_name": "basic1",
"completed": True,
"students": [{"name": "Abi"}, {"name": "Jack"}],
},
{
"course_name": "basic2",
"completed": True,
"students": [{"name": "Kate"}, {"name": "Miranda"}],
},
],
}
department = Department(**to_save)
await department.save_related(follow=True, save_all=True)
department_check = (
await Department.objects.select_all(follow=True)
.order_by(Department.courses.students.name.asc())
.get()
)
to_exclude = {
"id": ...,
"courses": {"id": ..., "students": {"id", "studentcourse"}},
}
assert department_check.dict(exclude=to_exclude) == to_save