feat: add aio_get_or_create

peewee_async.py

@@ -882,3 +882,24 @@ async def aio_create(cls, **query):
         inst = cls(**query)
         await inst.aio_save(force_insert=True)
         return inst
+
+    @classmethod
+    async def aio_get_or_create(cls, **kwargs):
+        defaults = kwargs.pop('defaults', {})
+        query = cls.select()
+        for field, value in kwargs.items():
+            query = query.where(getattr(cls, field) == value)
+
+        try:
+            return await query.aio_get(), False
+        except cls.DoesNotExist:
+            try:
+                if defaults:
+                    kwargs.update(defaults)
+                async with cls._meta.database.aio_atomic():
+                    return await cls.aio_create(**kwargs), True
+            except peewee.IntegrityError as exc:
+                try:
+                    return await query.aio_get(), False
+                except cls.DoesNotExist:
+                    raise exc

peewee_async_compat.py

@@ -265,6 +265,10 @@ async def get_or_create(self, model_, defaults=None, **kwargs):
         Return 2-tuple containing the model instance and a boolean
         indicating whether the instance was created.
         """
+        warnings.warn(
+            "`get_or_create` method is deprecated, use `AioModel.aio_get_or_create` instead.",
+            DeprecationWarning
+        )
         try:
             return (await self.get(model_, **kwargs)), False
         except model_.DoesNotExist:
@@ -342,6 +346,10 @@ async def create_or_get(self, model_, **kwargs):
         Try to create new object with specified data. If object already
         exists, then try to get it by unique fields.
         """
+        warnings.warn(
+            "`create_or_get` method is deprecated, use `AioModel.aio_get_or_create` instead.",
+            DeprecationWarning
+        )
         try:
             return (await self.create(model_, **kwargs)), True
         except IntegrityErrors:

tests/aio_model/test_shortcuts.py

@@ -109,3 +109,18 @@ async def test_aio_save__force_insert(db):
 
     with pytest.raises(peewee.IntegrityError):
         await t.aio_save(force_insert=True)
+
+
+@dbs_all
+async def test_aio_get_or_create__get(db):
+    t1 = await TestModel.aio_create(text="text", data="data")
+    t2, created = await TestModel.aio_get_or_create(text="text")
+    assert t1.id == t2.id
+    assert created is False
+
+
+@dbs_all
+async def test_aio_get_or_create__created(db):
+    t2, created = await TestModel.aio_get_or_create(text="text")
+    assert t2.text == "text"
+    assert created is True