Django Async CRUD - Basic CRUD operations with Django ORM in an async context

djang_async_crud.py

import typing

from django.db import models
from channels.db import database_sync_to_async


M = typing.TypeVar("M", bound=models.Model)


class AsyncCrud:
    def __init__(self, model: typing.Type[M]):
        self.model = model

    @database_sync_to_async
    def create(self, **kwargs) -> typing.Awaitable[M]:
        return self.model.objects.create(**kwargs)

    @database_sync_to_async
    def read(self, obj_id) -> typing.Awaitable[M]:
        return self.model.objects.get(id=obj_id)

    @database_sync_to_async
    def update(self, obj_id, **updates) -> typing.Awaitable[M]:
        obj = self.model.objects.get(id=obj_id)
        for attr, value in updates.items():
            setattr(obj, attr, value)
        obj.save()
        return obj

    @database_sync_to_async
    def delete(self, obj_id) -> typing.Awaitable[M]:
        obj = self.model.objects.get(id=obj_id)
        obj.delete()
        return obj

#
# usage
#

# Polls is a django model
PollsCrud = AsyncCrud(Polls)  

# C
poll = await PollsCrud.create(name="Are you ready to async?")

# R
print(await PollsCrud.read(poll.id)
   
# U      
await PollsCrud.update(poll.id, votes=5)

# D
await PollsCrud.delete(poll.id)