Try to create, if fails with IntegrityError then try to get
(
cls, db: BaseDBAsyncClient, defaults: dict, **kwargs
)
| 1293 | |
| 1294 | @classmethod |
| 1295 | async def _create_or_get( |
| 1296 | cls, db: BaseDBAsyncClient, defaults: dict, **kwargs |
| 1297 | ) -> tuple[Self, bool]: |
| 1298 | """Try to create, if fails with IntegrityError then try to get""" |
| 1299 | for key in defaults.keys() & kwargs.keys(): |
| 1300 | if (default_value := defaults[key]) != (query_value := kwargs[key]): |
| 1301 | raise ParamsError(f"Conflict value with {key=}: {default_value=} vs {query_value=}") |
| 1302 | merged_defaults = {**kwargs, **defaults} |
| 1303 | try: |
| 1304 | async with in_transaction(connection_name=db.connection_name) as connection: |
| 1305 | return await cls.create(using_db=connection, **merged_defaults), True |
| 1306 | except IntegrityError as exc: |
| 1307 | try: |
| 1308 | return await cls.filter(**kwargs).using_db(db).get(), False |
| 1309 | except DoesNotExist: |
| 1310 | pass |
| 1311 | raise exc |
| 1312 | |
| 1313 | @classmethod |
| 1314 | def _db_queryset( |
no test coverage detected