daviskirk · August 16, 2021 21:36
diff --git a/install.sh b/install.sh
 pip install fastapi celery[redis] uvicorn
diff --git a/main.py b/main.py
 import celery
 from fastapi import FastAPI
 import tasks

 app = FastAPI()

 @app.get("/")
 def read_root():
    result = tasks.add.delay(2, 2)
    return {"2+2": result.get()}
diff --git a/run.py b/run.py
 import uvicorn

 original_callback = uvicorn.main.callback

 def callback(**kwargs):
    from celery.contrib.testing.worker import start_worker
    import tasks

    with start_worker(tasks.app, perform_ping_check=False, loglevel="info"):
        original_callback(**kwargs)

 uvicorn.main.callback = callback


 if __name__ == "__main__":
    uvicorn.main()
diff --git a/run.sh b/run.sh
 python run.py main:app --reload
diff --git a/tasks.py b/tasks.py
 from celery import Celery

 app = Celery("tasks", broker='redis://localhost:6379/0', backend='redis://localhost')

 @app.task
 def add(x, y):
    return x + y
	import uvicorn

	original_callback = uvicorn.main.callback

	def callback(**kwargs):
	from celery.contrib.testing.worker import start_worker
	import tasks

	with start_worker(tasks.app, perform_ping_check=False, loglevel="info"):
	original_callback(**kwargs)

	uvicorn.main.callback = callback


	if __name__ == "__main__":
	uvicorn.main()
	from celery import Celery

	app = Celery("tasks", broker='redis://localhost:6379/0', backend='redis://localhost')

	@app.task
	def add(x, y):
	return x + y