implemented dockerfile and docker compose file

Author: Kenny Ng

.gitignore

@@ -157,4 +157,4 @@ cython_debug/
 #  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
+.idea/

README.md

@@ -1,8 +1,34 @@
-# celery-fastapi-integration
+# FastAPI Celery Integration Example
+
+## Forword
+I created this repository to show a barebone implementation of integrating FastAPI with Celery. 
+Unlike other repositories on GitHub demonstrating the same possibility, 
+flower is not used in this one as I wanted to make it as simple as it can be.
+
+In this example, rabbitmq is used as the message broker and redis is used as the result backend.
+
+Also, I added an example on how async function can be used on celery.
 
 ## Project Setup
-1. Run `pip install -r requirements` to install required libraries
-2. Run `docker run -d -p 5672:5672 rabbitmq` to run rabbitmq on docker, in which we will use it as the message broker
-3. Run `docker run -d -p 6379:6379 redis` to start redis, in which we use as the result backend for celery
-3. Run `celery -A tasks worker --loglevel=INFO` to start celery
-4. Run `uvicorn api.main:app --reload` to start FastAPI
+### Docker Build
+You can host the project by running the following commands:
+```
+# build image for celery
+docker build . -f worker.Dockerfile -t worker
+# build image for api
+docker build . -f api.Dockerfile -t api 
+# host celery, fastapi, rabbitmq and redis containers together
+docker-compose up -d --build
+```
+After hosting the containers on Docker, you can go to http://localhost:8000/docs to go the swagger UI or just call the API directly.
+
+## Usage
+By trigger the `POST run_task` endpoint, a task which takes 10 second will be trigger. A task ID will be issued for tracking progress and getting result.
+
+`GET status` can be used to track the status of a task with its ID.
+
+`GET result` can be used to get the result of the task.
+
+## Contact
+Please give the repo a star if you like it!
+This app is written by Kenny Ng ([email protected]).

api.Dockerfile

@@ -0,0 +1,11 @@
+FROM python:3.8-slim
+
+WORKDIR /app
+
+COPY requirements.txt ./
+
+RUN pip install -r requirements.txt
+
+COPY . .
+
+CMD ["uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "8000"]

api/main.py

@@ -1,14 +1,14 @@
 from typing import Union
-from task_queue import tasks
+from worker import tasks
 from celery.result import AsyncResult
 from fastapi import FastAPI
 from pydantic import BaseModel
-from task_queue.tasks import wait_10_sec
+from worker.tasks import example_task
 
 app = FastAPI()
 
 
-class InputDto(BaseModel):
+class ExampleInputModel(BaseModel):
     message: str = ''
 
 
@@ -17,14 +17,14 @@ def read_root():
     return {"Hello": "World"}
 
 
-@app.post("/run_task")
-def run_task(input_dto: InputDto):
+@app.post("/example_task")
+def run_task(input_dto: ExampleInputModel):
     json_body = input_dto.dict()
-    result = wait_10_sec.delay(json_body)
+    result = example_task.delay(json_body)
     return {'task_id': result.id}
 
 
-@app.get("/task_status")
+@app.get("/status")
 def get_task_status(task_id):
     task_result = AsyncResult(id=task_id)
     res = {
@@ -34,11 +34,11 @@ def get_task_status(task_id):
     return res
 
 
-@app.get("/tasK_result")
+@app.get("/result")
 def get_result(task_id):
     task_result = AsyncResult(task_id)
     result = {
         "id": task_id,
-        "task_result": task_result.result
+        "task_result": task_result.get()
     }
     return result

docker-compose.yml

@@ -0,0 +1,29 @@
+version: '3'
+
+services:
+  redis:
+    image: redis
+    ports:
+      - 6379:6379
+
+  rabbitmq:
+    image: rabbitmq
+    ports:
+      - 5672:5672
+      - 15672:15672
+
+  api:
+    image: api:latest
+    ports:
+      - 8000:8000
+    depends_on:
+      - rabbitmq
+      - redis
+
+  worker:
+    image: worker:latest
+    ports:
+      - 5555:5555
+    depends_on:
+      - rabbitmq
+      - redis

requirements.txt

@@ -1,4 +1,26 @@
-celery
-asyncio
-asgiref
-fastapi
+amqp==5.1.1
+anyio==3.6.2
+asgiref==3.6.0
+async-timeout==4.0.2
+asyncio==3.4.3
+billiard==3.6.4.0
+celery==5.2.7
+click==8.1.3
+click-didyoumean==0.3.0
+click-plugins==1.1.1
+click-repl==0.2.0
+fastapi==0.95.2
+h11==0.14.0
+idna==3.4
+kombu==5.2.4
+prompt-toolkit==3.0.38
+pydantic==1.10.7
+pytz==2023.3
+redis==4.5.5
+six==1.16.0
+sniffio==1.3.0
+starlette==0.27.0
+typing_extensions==4.5.0
+uvicorn==0.22.0
+vine==5.0.0
+wcwidth==0.2.6

task_queue/tasks.py

@@ -0,0 +1,11 @@
+FROM python:3.8-slim
+
+WORKDIR /app
+
+COPY requirements.txt ./
+
+RUN pip install -r requirements.txt
+
+COPY . .
+
+CMD celery -A worker.tasks worker --loglevel=INFO

test.py

@@ -0,0 +1,24 @@
+from celery import Celery
+from asgiref.sync import async_to_sync
+import asyncio
+
+# Initializing celery app
+app = Celery('tasks', backend='redis://redis:6379', broker='pyamqp://rabbitmq:5672')
+app.set_default()
+app.conf.update(task_track_started=True)
+app.conf.update(result_persistent=True)
+app.conf.update(worker_send_task_events=False)
+
+
+# A dummy async function which emulates the behavior of computations
+async def wait_10_sec_and_return(user_input):
+    await asyncio.sleep(10)
+    return user_input['message']
+
+
+@app.task
+def example_task(user_input):
+    # Turning async function into a sync one, as async functions are not supported in celery
+    async_service_to_sync = async_to_sync(wait_10_sec_and_return)
+    return async_service_to_sync(user_input)
+