Фоновая задача Django как FastAPI

У меня есть требование, что мне нужно вычислять хэш некоторых изображений в фоновом режиме после вставки данных в таблицы БД. Я видел реализацию фоновой задачи в FastAPI и Django, и она кажется очень простой и осуществимой в FastAPI по сравнению с Django. Поэтому мне нужно знать, есть ли подобная реализация на Django или нет? потому что это кажется слишком сложным на Django для простой фоновой задачи. Или лучше использовать простые multiprocessing.Process

Для Django: Мне нужно выполнить следующие шаги для достижения фонового задания

Установка

> cd [my-django-project root directory]
> pipenv install django-background-tasks

Теперь добавьте 'background_task' в INSTALLED_APPS в settings.py:

INSTALLED_APPS = (
    # ...
    'background_task',
    # ...
)

и выполнить миграцию базы данных, чтобы убедиться, что схема django-background-tasks находится на месте:

> pipenv shell
(my-django-project) bash-3.2$  python manage.py migrate

Создание и регистрация задачи

Любая функция Python может быть задачей, нам просто нужно применить аннотацию @background, чтобы зарегистрировать ее как таковую:

from background_task import background

@background(schedule=10)
def do_something(s1: str, s1: str) -> None:
   """
   Does something that takes a long time
   :param p1: first parameter
   :param p2: second parameter
   :return: None
   """
   pass

Теперь мы можем вызывать функцию как обычно в нашем проекте:

do_something("first parameter", "second parameter")

Важно отметить, что вызов функции не фактически выполняет ее код, а запись о задаче сохраняется в базе данных модулем "django-background-tasks", точнее в таблице "background_task". По этой причине написание функции задачи, которая что-то возвращает, не имеет смысла, поскольку задача все равно будет выполнена в фоновом режиме в более поздний момент, поэтому "значение", возвращаемое функцией в момент ее вызова, практически бессмысленно. Единственный случай использования возвращаемого значения, который я вижу, это тестирование, см. раздел Тестирование задачи ниже.

Задачи обработки

Для того чтобы действительно запустить зарегистрированную задачу, мы должны использовать следующую команду управления:

> python manage.py process_tasks 

Для FastAPI:

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
      content = f"notification for {email}: {message}"
    email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, bg_tsk: BackgroundTasks):

bg_tsk.add_task(write_notification, email, message="some notification")
return {"message": "Notification sent in the background"}

Вернуться на верх