Repository: tortoise/tortoise-orm

Issue:
IntegrityError with unique field and get_or_create
**Describe the bug**
Hi!
First of all thank you for developing such useful ORM.
I am not sure that this is really bug, maybe I am doing something wrong.
I get IntegrityError with fields having unique=True and using get_or_create.

**To Reproduce**
Models.py is:

    class CheckDates(Model):
        id = fields.IntField(pk=True)
        check_date = fields.DateField(unique=True)

        def __str__(self):
            return self.check_date.strftime('%Y-%m-%d')

    class Projects(Model):
        id = fields.IntField(pk=True)
        name = fields.CharField(max_length=255, unique=True)

        def __str__(self):
            return self.name

    class Results(Model):
        id = fields.IntField(pk=True)
        date = fields.ForeignKeyField('models.CheckDates', related_name='results')
        project = fields.ForeignKeyField('models.Projects', related_name='results')
        suspicious_url = fields.CharField(max_length=255)

        class Meta:
            unique_together = ('date', 'project', 'suspicious_url')

        def __str__(self):
            return self.suspicious_url


I have file for generating schemas:

    from tortoise import Tortoise, run_async


    async def init(generate=False):
        await Tortoise.init(
            db_url='sqlite://db.sqlite3',
            modules={'models': ['AntiDoorway.models']}
        )
        if generate:
            await generate_schemas()

    async def generate_schemas():
        await Tortoise.generate_schemas()

    async def close_conns():
        await Tortoise.close_connections()

    if __name__ == '__main__':
        run_async(init(generate=True))


And file with main pipeline (some code is excluded as irrelevant):
    
    def check_by_visiting(urls_dict: dict, project_name: str, check_date: date, counter_id: int) -> None:
        tasks = []
        loop = asyncio.get_event_loop()
        loop.run_until_complete(init())
        for url in urls_dict.keys():
            for platform in platforms:
                tasks.append(loop.create_task(main(url, project_name, check_date, counter_id, platform)))
        loop.run_until_complete(asyncio.wait(tasks))
        loop.run_until_complete(close_conns())
        loop.close()

    async def main(url: str, project_name: str, check_date: date, counter_id: int, platform: str) -> None:
        result = []
        try:
            date_id = await CheckDates.get_or_create(check_date=check_date)
        except exceptions.IntegrityError:
            date_id = await CheckDates.filter(check_date=check_date)
        date_id = date_id[0]
        project_id = await Projects.get_or_create(name=project_name)
        project_id = project_id[0]

        params = await form_request_params(platform)
        formed_url = await form_url(url)
        async with aiohttp.ClientSession() as session:
            try:
                response = await fetch_data(formed_url, params, session)
            except Exception as e:
                response = ''
        result.append(await check_counter_presence(response, counter_id))
        result.append(await return_reload_in_header(response))
        if any(result):
            try:
                await Results.get_or_create(date=date_id, project=project_id, suspicious_url=url)
            except (sqlite3.IntegrityError, exceptions.IntegrityError):
                pass

I get following errors when I run script with empty DB:
sqlite3.IntegrityError: UNIQUE constraint failed: checkdates.check_date
and
sqlite3.IntegrityError: UNIQUE constraint failed: results.date_id, results.project_id, results.suspicious_url
but all data is saved (well, I hope it is).
And no errors if date is already in DB.

**Expected behavior**
No error messages.

**Additional context**
So this might be a bug if I'm doing it right.

Discussion:
grigi: @abondar based this on your original attempt to fix this. Please review.
coveralls: ## Pull Request Test Coverage Report for [Build 1330](https://coveralls.io/builds/29025873)

* **27** of **28**   **(96.43%)**  changed or added relevant lines in **4** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage decreased (**-0.1%**) to **97.959%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [tortoise/models.py](https://coveralls.io/builds/29025873/source?filename=tortoise%2Fmodels.py#L766) | 13 | 14 | 92.86%
<!-- | **Total:** | **27** | **28** | **96.43%** | -->


|  Totals | [![Coverage Status](https://coveralls.io/builds/29025873/badge)](https://coveralls.io/builds/29025873) |
| :-- | --: |
| Change from base [Build 1324](https://coveralls.io/builds/28992811): |  -0.1% |
| Covered Lines: | 3516 |
| Relevant Lines: | 3551 |

---
##### 💛  - [Coveralls](https://coveralls.io)

grigi: Wow, Py3.6 gets so confused with contexts :sob: 
I feel we may just have to remove it from CI altogether...
It has been deprecated for a few months now...

Unfortunately it would mean dropping pypy too.

I'll just conditionally skip the test for now.
grigi: Another option is to disable all connection pooling on py3.6?
AEnterprise: wow didn't expect pooling to be such a major pain on 3.6

but personally i'm more in favor for dropping only specific features like pooling then dropping an entire version (prob best to log a warning on startup though so people are aware)
grigi: Py3.6 uses a contextvar backport, and unfortunately it has shortcomings. py3.7+ has extra state in their asyncio C-structs to keep track better.
I'm busy seeing what would be the lowest-impact change for disabling pooling for py3.6. It seems forcing a pool of 1 doesn't do it.
I suspect I would have to use explicit locking like I do for SQLite to get it working right.

AEnterprise: i think forcing a shared lock instead of making new locks for concurrent operations should do the trick

that and forcing a poolsize of 1 regardless since it can't use multiple so there is no reason to have any extras open
grigi: Well, in the meantime I'm getting hillarious errors, like "cannot rollback; the transaction is not yet started" but investigating further the cause is the dreaded "cannot perform operation: another operation is in progress".

And fixing race conditions fixes that. 
grigi: @AEnterprise We are discussing the potential deprecation for py3.6 in #303
This PR fixes a bug, and should go in soon.
I'll merge this to master instead of develop, and cherry-pick the windows fix as well and release that. All the new feature work is going in as v0.16
grigi: I'll cherry pick the commits instead.

Patch:
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 19d7564..7f9034b 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -74,6 +74,7 @@ future
 
 - Fix default type of ``JSONField``
 - Install on Windows does not require a C compiler any more.
+- Fix ``IntegrityError`` with unique field and ``get_or_create``
 
 0.15.17
 -------
diff --git a/tortoise/backends/asyncpg/client.py b/tortoise/backends/asyncpg/client.py
index 470c7db..d78f329 100644
--- a/tortoise/backends/asyncpg/client.py
+++ b/tortoise/backends/asyncpg/client.py
@@ -13,7 +13,7 @@ from tortoise.backends.base.client import (
     BaseTransactionWrapper,
     Capabilities,
     ConnectionWrapper,
-    NestedTransactionContext,
+    NestedTransactionPooledContext,
     PoolConnectionWrapper,
     TransactionContext,
     TransactionContextPooled,
@@ -192,6 +192,7 @@ class TransactionWrapper(AsyncpgDBClient, BaseTransactionWrapper):
     def __init__(self, connection: AsyncpgDBClient) -> None:
         self._connection: asyncpg.Connection = connection._connection
         self._lock = asyncio.Lock()
+        self._trxlock = asyncio.Lock()
         self.log = connection.log
         self.connection_name = connection.connection_name
         self.transaction: Transaction = None
@@ -199,7 +200,7 @@ class TransactionWrapper(AsyncpgDBClient, BaseTransactionWrapper):
         self._parent = connection
 
     def _in_transaction(self) -> "TransactionContext":
-        return NestedTransactionContext(self)
+        return NestedTransactionPooledContext(self)
 
     def acquire_connection(self) -> "ConnectionWrapper":
         return ConnectionWrapper(self._connection, self._lock)
diff --git a/tortoise/backends/base/client.py b/tortoise/backends/base/client.py
index bb1fd72..69ed9ca 100644
--- a/tortoise/backends/base/client.py
+++ b/tortoise/backends/base/client.py
@@ -268,7 +268,6 @@ class TransactionContextPooled(TransactionContext):
 
 class NestedTransactionContext(TransactionContext):
     async def __aenter__(self):
-        await self.connection.start()
         return self.connection
 
     async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
@@ -277,8 +276,20 @@ class NestedTransactionContext(TransactionContext):
                 # Can't rollback a transaction that already failed.
                 if exc_type is not TransactionManagementError:
                     await self.connection.rollback()
-            else:
-                await self.connection.commit()
+
+
+class NestedTransactionPooledContext(TransactionContext):
+    async def __aenter__(self):
+        await self.lock.acquire()
+        return self.connection
+
+    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        self.lock.release()
+        if not self.connection._finalized:
+            if exc_type:
+                # Can't rollback a transaction that already failed.
+                if exc_type is not TransactionManagementError:
+                    await self.connection.rollback()
 
 
 class PoolConnectionWrapper:
diff --git a/tortoise/backends/mysql/client.py b/tortoise/backends/mysql/client.py
index ecb7ca4..49963c5 100644
--- a/tortoise/backends/mysql/client.py
+++ b/tortoise/backends/mysql/client.py
@@ -12,7 +12,7 @@ from tortoise.backends.base.client import (
     BaseTransactionWrapper,
     Capabilities,
     ConnectionWrapper,
-    NestedTransactionContext,
+    NestedTransactionPooledContext,
     PoolConnectionWrapper,
     TransactionContext,
     TransactionContextPooled,
@@ -208,13 +208,14 @@ class TransactionWrapper(MySQLClient, BaseTransactionWrapper):
         self.connection_name = connection.connection_name
         self._connection: aiomysql.Connection = connection._connection
         self._lock = asyncio.Lock()
+        self._trxlock = asyncio.Lock()
         self.log = connection.log
         self._finalized: Optional[bool] = None
         self.fetch_inserted = connection.fetch_inserted
         self._parent = connection
 
     def _in_transaction(self) -> "TransactionContext":
-        return NestedTransactionContext(self)
+        return NestedTransactionPooledContext(self)
 
     def acquire_connection(self) -> ConnectionWrapper:
         return ConnectionWrapper(self._connection, self._lock)
diff --git a/tortoise/models.py b/tortoise/models.py
index 377a001..b5169df 100644
--- a/tortoise/models.py
+++ b/tortoise/models.py
@@ -7,7 +7,12 @@ from typing import Any, Awaitable, Dict, Generator, List, Optional, Set, Tuple,
 from pypika import Order, Query, Table
 
 from tortoise.backends.base.client import BaseDBAsyncClient
-from tortoise.exceptions import ConfigurationError, OperationalError
+from tortoise.exceptions import (
+    ConfigurationError,
+    IntegrityError,
+    OperationalError,
+    TransactionManagementError,
+)
 from tortoise.fields.base import Field
 from tortoise.fields.data import IntField
 from tortoise.fields.relational import (
@@ -22,7 +27,7 @@ from tortoise.fields.relational import (
 from tortoise.filters import get_filters_for_field
 from tortoise.functions import Function
 from tortoise.queryset import Q, QuerySet, QuerySetSingle
-from tortoise.transactions import current_transaction_map
+from tortoise.transactions import current_transaction_map, in_transaction
 
 MODEL = TypeVar("MODEL", bound="Model")
 # TODO: Define Filter type object. Possibly tuple?
@@ -752,10 +757,18 @@ class Model(metaclass=ModelMeta):
         """
         if not defaults:
             defaults = {}
-        instance = await cls.filter(**kwargs).first()
-        if instance:
-            return instance, False
-        return await cls.create(**defaults, **kwargs, using_db=using_db), True
+        db = using_db if using_db else cls._meta.db
+        async with in_transaction(connection_name=db.connection_name):
+            instance = await cls.filter(**kwargs).first()
+            if instance:
+                return instance, False
+            try:
+                return await cls.create(**defaults, **kwargs), True
+            except (IntegrityError, TransactionManagementError):
+                # Let transaction close
+                pass
+        # Try after transaction in case transaction error
+        return await cls.get(**kwargs), False
 
     @classmethod
     async def create(cls: Type[MODEL], **kwargs: Any) -> MODEL:


Test patch:
diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py
index c6bdc37..405491e 100644
--- a/tests/test_concurrency.py
+++ b/tests/test_concurrency.py
@@ -29,7 +29,7 @@ class TestConcurrencyIsolated(test.IsolatedTestCase):
         for una in unas:
             self.assertEqual(una[0], unas[0][0])
 
-    @test.expectedFailure
+    @test.skipIf(sys.version_info < (3, 7), "aiocontextvars backport not handling this well")
     async def test_concurrent_get_or_create(self):
         unas = await asyncio.gather(*[UniqueName.get_or_create(name="d") for _ in range(10)])
         una_created = [una[1] for una in unas if una[1] is True]
@@ -76,7 +76,7 @@ class TestConcurrencyTransactioned(test.TestCase):
         for una in unas:
             self.assertEqual(una[0], unas[0][0])
 
-    @test.skip("Crashes with MySQL & PostgreSQL")
+    @test.skipIf(sys.version_info < (3, 7), "aiocontextvars backport not handling this well")
     async def test_concurrent_get_or_create(self):
         unas = await asyncio.gather(*[UniqueName.get_or_create(name="b") for _ in range(10)])
         una_created = [una[1] for una in unas if una[1] is True]
diff --git a/tests/test_queryset.py b/tests/test_queryset.py
index 348bccf..3dd23c2 100644
--- a/tests/test_queryset.py
+++ b/tests/test_queryset.py
@@ -1,5 +1,3 @@
-import sys
-
 from tests.testmodels import IntFields, MinRelation, Tournament
 from tortoise.contrib import test
 from tortoise.exceptions import DoesNotExist, FieldError, IntegrityError, MultipleObjectsReturned
@@ -278,7 +276,6 @@ class TestQueryset(test.TestCase):
         ):
             await IntFields.all().values_list(flat=True)
 
-    @test.skipIf(sys.version_info < (3, 6), "Class fields not sorted in 3.5")
     async def test_all_values_list(self):
         data = await IntFields.all().order_by("id").values_list()
         self.assertEqual(data[2], (self.intfields[2].id, 16, None))