Add celery tasks

Author: jonra1993

Diff for: Makefile

@@ -79,6 +79,12 @@ run-prod:
 stop-prod:
 	docker compose down
 
+create-celery-db:
+	if ! docker compose -f docker-compose-dev.yml exec database psql -U ${DATABASE_USER} -h localhost -lqt | cut -d \| -f 1 | grep -qw ${DATABASE_CELERY_NAME}; then \
+		docker compose -f docker-compose-dev.yml exec database createdb -U ${DATABASE_USER} -W ${DATABASE_PASSWORD} -h localhost -O ${DATABASE_USER} ${DATABASE_CELERY_NAME}; \
+	fi
+	
+
 init-db:
 	docker compose -f docker-compose-dev.yml exec fastapi_server python app/initial_data.py && \
 	echo "Initial data created." 

Diff for: backend/app/app/api/celery_task.py

@@ -1,9 +1,31 @@
+import asyncio
 import time
+from uuid import UUID
+from app import crud
 from app.core.celery import celery
+from app.models.hero_model import Hero
+from app.schemas.hero_schema import IHeroCreate, IHeroRead, IHeroReadWithTeam
+from app.db.session import SessionLocal
+from asyncer import runnify
 
 @celery.task(name="increment_a_value")
 def increment(value: int) -> int:
     time.sleep(5)
     new_value = value + 1
     print("new_value", new_value)
     return new_value
+
+
+async def get_hero(hero_id: UUID) -> Hero:
+    async with SessionLocal() as session:
+        await asyncio.sleep(5) # Add a delay of 5 seconds
+        hero = await crud.hero.get(id=hero_id, db_session=session)
+        return hero
+
+
+@celery.task(name="print_hero")
+def print_hero(hero_id: UUID) -> None:
+    hero = runnify(get_hero)(hero_id=hero_id)
+    print(f"hero_id {hero.id}") 
+    return hero.id
+

Diff for: backend/app/app/api/v1/endpoints/hero.py

@@ -1,4 +1,5 @@
 from uuid import UUID
+from app.api.celery_task import print_hero
 from app.utils.exceptions import IdNotFoundException, NameNotFoundException
 from fastapi import APIRouter, Depends, HTTPException, Query
 from fastapi_pagination import Params
@@ -23,6 +24,7 @@
 )
 from app.schemas.role_schema import IRoleEnum
 from app.core.authz import is_authorized
+from fastapi.encoders import jsonable_encoder
 
 router = APIRouter()
 
@@ -68,6 +70,8 @@ async def get_hero_by_id(
     hero = await crud.hero.get(id=hero_id)
     if not hero:
         raise IdNotFoundException(Hero, hero_id)
+    
+    print_hero.delay(hero.id)
     return create_response(data=hero)
 
 

Diff for: backend/app/app/deps/celery_deps.py

@@ -0,0 +1,12 @@
+from typing import Generator
+from celery_sqlalchemy_scheduler.session import SessionManager
+from app.core.config import settings
+
+def get_job_db() ->Generator:
+    try:
+        session_manager = SessionManager()
+        engine, _session = session_manager.create_session(settings.SYNC_CELERY_BEAT_DATABASE_URI)
+        session = _session()
+        return session
+    finally:
+        session.close()

Diff for: backend/app/app/main.py

@@ -1,7 +1,9 @@
 from datetime import datetime, timedelta
 import gc
+import json
 from typing import Any
-from fastapi import FastAPI, HTTPException
+from app.deps.celery_deps import get_job_db
+from fastapi import Depends, FastAPI, HTTPException
 from app.api.deps import get_redis_client
 from fastapi_pagination import add_pagination
 from starlette.middleware.cors import CORSMiddleware
@@ -15,6 +17,13 @@
 from transformers import pipeline
 from app.api.celery_task import increment
 from app.core.celery import celery
+from celery_sqlalchemy_scheduler.models import (
+    PeriodicTask,
+    IntervalSchedule,
+    CrontabSchedule,
+)
+from sqlmodel import select
+
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
@@ -49,7 +58,6 @@ async def lifespan(app: FastAPI):
 )
 
 
-
 app.add_middleware(
     SQLAlchemyMiddleware,
     db_url=settings.ASYNC_DATABASE_URI,
@@ -90,18 +98,16 @@ async def root():
     An example "Hello world" FastAPI route.
     """
 
-
-    val = increment.delay(1)  #wait
-    increment.delay(9)  #no wait
+    val = increment.delay(1)  # wait
+    increment.delay(9)  # no wait
     tomorrow = datetime.utcnow() + timedelta(seconds=20)
-    new_tomorrow = increment.apply_async(args=[7], eta=tomorrow)
+    new_tomorrow = increment.apply_async(args=[8], eta=tomorrow)
     print("task_id", new_tomorrow.task_id)
     print("result", new_tomorrow.result)
     print("status", new_tomorrow.status)
     increment.apply_async(args=[20], expires=datetime.now() + timedelta(seconds=10))
-    
-    
-    # if oso.is_allowed(user, "read", message):    
+
+    # if oso.is_allowed(user, "read", message):
     return {"message": new_tomorrow.task_id}
 
 
@@ -112,16 +118,93 @@ async def root(task_id: Any):
     """
     # Retrieve the result using the task ID
     async_result = celery.AsyncResult(task_id)
-    print("state", async_result.state)
-    print("ready", async_result.ready())
-    print("successful", async_result.successful())
+
     if async_result.ready():
-        print(f"Task {task_id} exists and has completed.")
+        if not async_result.successful():
+            raise HTTPException(
+                status_code=404,
+                detail=f"Task {task_id} with state {async_result.state}.",
+            )
+
         result = async_result.get(timeout=1.0)
         return {"message": result}
-    else:        
-        raise HTTPException(status_code=404, detail="Task {task_id} does not exist or is still running.")
+    else:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Task {task_id} does not exist or is still running.",
+        )
+
+
+@app.get("/3")
+async def root(celery_session=Depends(get_job_db)):
+    """
+    An example "Hello world" FastAPI route.
+    """
+    # Retrieve the result using the task ID
+    periodic_task = PeriodicTask(
+        crontab=CrontabSchedule(
+            hour=14, minute=47, day_of_month=24, month_of_year=3, timezone="UTC"
+        ),
+        name="new_uuid",
+        args="[8]",
+        task="increment_a_value",
+        one_off=True,
+    )
+    celery_session.add(periodic_task)
+    celery_session.commit()
+    celery_session.close()
+
+    return {"message": "hello"}
+
+
+@app.get("/4")
+async def root(celery_session=Depends(get_job_db)):
+    """
+    An example "Hello world" FastAPI route.
+    """
+    # Retrieve the result using the task ID
+    query = select(PeriodicTask).where(PeriodicTask.name == "new_uuid")
+    periodic_task = celery_session.execute(query).scalar_one_or_none()
 
+    periodic_task.crontab = CrontabSchedule(
+            hour=19, minute=58, day_of_month=24, month_of_year=3, timezone="UTC"
+        )
+    celery_session.add(periodic_task)
+    celery_session.commit()
+    celery_session.close()
+
+    return {"message": "hello"}
+
+@app.get("/5")
+async def root(celery_session=Depends(get_job_db)):
+    """
+    An example "Hello world" FastAPI route.
+    """
+    # Retrieve the result using the task ID
+    periodic_task = PeriodicTask(
+        interval=IntervalSchedule(every=10, period=IntervalSchedule.SECONDS),
+        name="new_uuid_new_interval",
+        args="[8]",
+        task="increment_a_value",
+    )
+    celery_session.add(periodic_task)
+    celery_session.commit()
+    celery_session.close()
+
+@app.get("/6")
+async def root(celery_session=Depends(get_job_db)):
+    """
+    An example "Hello world" FastAPI route.
+    """
+    # Retrieve the result using the task ID
+    query = select(PeriodicTask).where(PeriodicTask.name == "new_uuid_new_interval")
+    periodic_task = celery_session.execute(query).scalar_one_or_none()
+    periodic_task.enabled = False
+    celery_session.add(periodic_task)
+    celery_session.commit()
+    celery_session.close()
+
+    return {"message": "hello"}
 
 # Add Routers
 app.include_router(api_router_v1, prefix=settings.API_V1_STR)

Diff for: backend/app/poetry.lock

@@ -58,6 +58,7 @@ transformers = {extras = ["torch"], version = "^4.27.2"}
 celery = "^5.2.7"
 celery-sqlalchemy-scheduler = "^0.3.0"
 psycopg2-binary = "^2.9.5"
+watchdog = "^3.0.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^5.2"

Diff for: backend/app/pyproject.toml

@@ -0,0 +1,2 @@
+CREATE DATABASE fastapi_db;
+CREATE DATABASE celery_schedule_jobs;

Diff for: create-dbs.sql

@@ -24,24 +24,44 @@ services:
       - "traefik.http.routers.fastapi_local_base.rule=Host(`${LOCAL_1}`, `${LOCAL_2}`, `${EXT_ENDPOINT1}`)"
       - traefik.http.services.fastapi_localservice.loadbalancer.server.port=8000
 
+  # database:
+  #   image: postgres
+  #   restart: always
+  #   container_name: database
+  #   env_file: ".env"
+  #   #user: root
+  #   volumes:
+  #     - ./db_docker:/var/lib/postgresql/data
+  #     #- ./db_docker:/bitnami/postgresql
+  #     - ./init-multi-postgres-databases.sh:/docker-entrypoint-initdb.d/init-multi-postgres-databases.sh
+  #   ports:
+  #     - 5454:5432  # Remove this on production
+  #   expose:
+  #     - 5432
+  #   environment:
+  #     - POSTGRES_USERNAME=${DATABASE_USER}
+  #     - POSTGRES_PASSWORD=${DATABASE_PASSWORD}
+  #     - POSTGRES_DB=fastapi_db
+  #     #- POSTGRES_DATABASE=${DATABASE_NAME},${DATABASE_CELERY_NAME}
+  #     - POSTGRES_MULTIBLE_DATABASES=fastapi_db,celery_schedule_jobs
+  #   #command: ["bash", "-c", "chmod +x /docker-entrypoint-initdb.d/init-multi-postgres-databases.sh && postgres"]
+
   database:
-    image: bitnami/postgresql:13.3.0
+    image: postgres:latest
     restart: always
     container_name: database
     env_file: ".env"
-    user: root
-    volumes:
-      - ./db_docker:/bitnami/postgresql
     ports:
       - 5454:5432  # Remove this on production
     expose:
       - 5432
+    volumes:
+      - ./db_docker:/var/lib/postgresql/data
+      - ./create-dbs.sql:/docker-entrypoint-initdb.d/create-dbs.sql
     environment:
-      - POSTGRES_USERNAME=${DATABASE_USER}
+      - POSTGRES_USER=${DATABASE_USER}
       - POSTGRES_PASSWORD=${DATABASE_PASSWORD}
-      - POSTGRES_DATABASE=${DATABASE_NAME}
-      - POSTGRES_HOST_AUTH_METHOD= "trust"
-      - POSTGRESQL_POSTGRES_PASSWORD= ${DATABASE_PASSWORD}
+    
 
   redis_server:
     container_name: redis_server
@@ -59,9 +79,7 @@ services:
       context: ./backend      
       args:
         INSTALL_DEV: ${INSTALL_DEV-false}
-    command: "celery -A app.core.celery worker -l info "
-    # celery -A my_celery_app worker --without-heartbeat --without-gossip --without-mingle
-    # command: " watchmedo auto-restart --directory=./ --pattern=*.py --recursive -- celery -A app.core.worker worker -l info "
+    command: "watchfiles 'celery -A app.core.celery worker -l info' "
     volumes:
       - ./backend/app:/code
       # - "${EB_LOG_BASE_DIR}/php-app:/var/log/celery"
@@ -86,6 +104,7 @@ services:
       - database
       - redis_server
     env_file: .env
+  
 
   nginx_server:
     container_name: nginx_server