Django で async のメソッドから ORM を実行する例。
model の実装例
以下の Book、Article、Author を定義。
test1/models.py
from django.db import models
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
class Book(models.Model):
title = models.CharField(max_length=128)
def __str__(self):
return self.title
class Article(models.Model):
headline = models.CharField(max_length=128)
def __str__(self):
return self.headline
class Author(models.Model):
name = models.CharField(max_length=128)
target_id = models.PositiveIntegerField()
target_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
target = GenericForeignKey('target_type', 'target_id')
def __str__(self):
return self.name
async メソッドから Django ORM を実行
async のメソッドから Django ORM を実行するとエラーになる。
test1/management/commands/select_async_err.py
import asyncio
from django.core.management.base import BaseCommand
from django.contrib.contenttypes.models import ContentType
from test1.models import Book, Article, Author
class Command(BaseCommand):
def get_records(self, str):
print(f"get_records(): {str}")
book_target_type = ContentType.objects.get_for_model(Book)
book_authors = Author.objects.filter(target_type=book_target_type)
print(book_authors)
book_authors_sorted = book_authors.order_by('-name')
print(book_authors_sorted)
return book_authors_sorted
async def async_func_error(self, str):
print(f"async_func_error()")
res = self.get_records(str)
print(f"res: {res}")
def handle(self, *args, **options):
asyncio.run(self.async_func_error("err"))
実行
python manage.py async_err
実行結果
raise SynchronousOnlyOperation(message)
django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.
sync_to_async() で Django ORM を実行
sync_to_async() を使って Django ORM のメソッドの呼び出し方を変更する。
test1/management/commands/select_async.py
import asyncio
from asgiref.sync import sync_to_async
from django.core.management.base import BaseCommand
from django.contrib.contenttypes.models import ContentType
from test1.models import Book, Article, Author
class Command(BaseCommand):
def get_records(self, str):
print(f"get_records(): {str}")
book_target_type = ContentType.objects.get_for_model(Book)
book_authors = Author.objects.filter(target_type=book_target_type)
print(book_authors)
book_authors_sorted = book_authors.order_by('-name')
print(book_authors_sorted)
return book_authors_sorted
async def async_func(self, str):
print(f"async_func()")
async_task = sync_to_async(self.get_records)(str)
task = asyncio.create_task(async_task)
res = task
print(f"res: {res}")
def handle(self, *args, **options):
asyncio.run(self.async_func("abc"))
実行
python manage.py select_async
実行結果
async_func()
res: <Task pending name='Task-2' coro=<SyncToAsync.__call__() running at ...>>
get_records(): abc
<QuerySet [<Author: book1_author>, <Author: book2_author>]>
<QuerySet [<Author: book2_author>, <Author: book1_author>]>