setup celery for email send task

Author: dfm88

docker/Dockerfile

@@ -75,6 +75,6 @@ ENV PATH /${IMAGE_NAME}/venv/bin:$PATH
 WORKDIR /${IMAGE_NAME}
 COPY . ./
 
-RUN chmod +x /${IMAGE_NAME}/docker/entrypoint.sh
+RUN chmod +x /${IMAGE_NAME}/scripts/entrypoint.sh
 
-ENTRYPOINT ["sh", "/docker/entrypoint.sh"]
+ENTRYPOINT ["sh", "scripts/entrypoint.sh"]

docker/docker-compose.yaml

@@ -13,10 +13,13 @@ services:
       - ./data/pgdata:/var/lib/postgresql/my_data
     env_file:
       - ../env/.env
+    environment:
+      - SQLALCHEMY_DATABASE_URI=postgresql+asyncpg://admin:admin@fastapi-2fa-db:5432/postgres
+      - CELERY_BROKER_URL=amqp://admin:admin@fastapi-2fa-rabbitmq:5672/
+      - result_backend=redis://fastapi-2fa-cache:6379/0
+
     ports:
       - "5454:5432"
-    networks:
-      - servicenetwork
 
 ## REDIS
   redis:
@@ -25,11 +28,62 @@ services:
     restart: always
     env_file:
       - ../env/.env
+    environment:
+      - SQLALCHEMY_DATABASE_URI=postgresql+asyncpg://admin:admin@fastapi-2fa-db:5432/postgres
+      - CELERY_BROKER_URL=amqp://admin:admin@fastapi-2fa-rabbitmq:5672/
+      - result_backend=redis://fastapi-2fa-cache:6379/0
     ports:
       - "6389:6379"
-    networks:
-      - servicenetwork
 
-networks:
-    servicenetwork:
-        driver: bridge
+  rabbitmq:
+    container_name: fastapi-2fa-rabbitmq
+    image: rabbitmq:3-management
+    env_file:
+      - ../env/.env
+    environment:
+      - SQLALCHEMY_DATABASE_URI=postgresql+asyncpg://admin:admin@fastapi-2fa-db:5432/postgres
+      - CELERY_BROKER_URL=amqp://admin:admin@fastapi-2fa-rabbitmq:5672/
+      - result_backend=redis://fastapi-2fa-cache:6379/0
+    ports:
+      - "5672:5672"
+
+  celery_worker:
+    container_name: fastapi-2fa-celery
+    build:
+      context: ..
+      dockerfile: ./docker/Dockerfile
+    image: fastapi_2fa_celery_worker
+    entrypoint: /scripts/start_celery_worker.sh
+    env_file:
+      - ../env/.env
+    environment:
+      - SQLALCHEMY_DATABASE_URI=postgresql+asyncpg://admin:admin@fastapi-2fa-db:5432/postgres
+      - CELERY_BROKER_URL=amqp://admin:admin@fastapi-2fa-rabbitmq:5672/
+      - result_backend=redis://fastapi-2fa-cache:6379/0
+    depends_on:
+        - redis
+        - db
+        - rabbitmq
+
+  flower:
+    container_name: fastapi-2fa-flower
+    build:
+      context: ..
+      dockerfile: ./docker/Dockerfile
+    image: fastapi_2fa_celery_flower
+    entrypoint: /scripts/start_celery_flower.sh
+    volumes:
+        - flower_db:/app/flower_db
+    env_file:
+      - ../env/.env
+    ports:
+      - 5557:5557
+    depends_on:
+        - redis
+        - db
+        - rabbitmq
+        - celery_worker
+
+volumes:
+   pgdata:
+   flower_db:

env/.env

@@ -1,15 +1,17 @@
 FASTAPI_CONFIG=development
-
 FAKE_EMAIL_SENDER=[email protected]
-
 API_V1_STR=/api/v1
 PROJECT_NAME=fastapi-2fa
 
+# # JWT
 JWT_SECRET_KEY=9d58ca3446806034a68b02cbcdf69d8f
 JWT_SECRET_KEY_REFRESH=mXuqbFTs2kTmz+6rBpJh1B4T+zHQh8Nq1nc7BOrWTb4=
 PRE_TFA_SECRET_KEY=11rWUgZLTckPHI7KO1SPbFgG1OILrDQjI7v9Q7KgYZw=
 ALGORITHM=HS256
+ACCESS_TOKEN_EXPIRE_MINUTES=30
+REFRESH_TOKEN_EXPIRE_MINUTES=1440  # 24 h
 
+# # 2 FACTOR AUTHENTICATION
 FERNET_KEY_TFA_TOKEN=J_TYpprFmoLlVM0MNZElt8IwEkvEEhAwCmb8P_f7Fro=
 TFA_BACKUP_TOKENS_NR=5
 TFA_TOKEN_LENGTH=6
@@ -18,16 +20,27 @@ TFA_TOKEN_LENGTH=6
 # -->MAX = 10 => 5 minutes
 TOTP_TOKEN_TOLERANCE=2
 TOTP_ISSUER_NAME=fastapi_2fa
-
-ACCESS_TOKEN_EXPIRE_MINUTES=30
-REFRESH_TOKEN_EXPIRE_MINUTES=1440  # 24 h
 PRE_TFA_TOKEN_EXPIRE_MINUTES=5
 
+# # CORS
 BACKEND_CORS_ORIGINS=http://localhost:5555
 
+# # DB
 POSTGRES_USER=admin
 POSTGRES_PASSWORD=admin
 POSTGRES_HOST=localhost
-POSTGRES_PORT=5430
+POSTGRES_PORT=5432
 POSTGRES_DB=postgres
-SQLALCHEMY_DATABASE_URI=postgresql+asyncpg://admin:admin@localhost:5454/postgres
+SQLALCHEMY_DATABASE_URI=postgresql+asyncpg://admin:admin@localhost:5454/postgres
+
+# # RABBIT MQ
+RABBITMQ_DEFAULT_USER=admin
+RABBITMQ_DEFAULT_PASS=admin
+
+# # CELERY
+CELERY_BROKER_URL=amqp://admin:admin@localhost:5672/
+result_backend=redis://localhost:6389/0
+
+# # FLOWER
+CELERY_FLOWER_USER=admin
+CELERY_FLOWER_PASSWORD=admin

fastapi_2fa/api/endpoints/api_v1/tasks.py

@@ -0,0 +1,45 @@
+from celery.result import AsyncResult
+from fastapi import APIRouter
+
+from fastapi_2fa.core.config import settings
+from fastapi_2fa.core.utils import Email
+from fastapi_2fa.tasks.tasks import send_email_task
+
+tasks_router = APIRouter()
+
+
+@tasks_router.get(
+    "/test-celery",
+    description="endpoint to test celery send mail function"
+)
+async def testcelery():
+    """
+    Test celery
+    """
+    email_ob = Email(
+        to_=[settings.FAKE_EMAIL_SENDER],
+        from_=settings.FAKE_EMAIL_SENDER,
+        text_="Prova messaggio con Celery"
+    )
+    task = send_email_task.apply_async(kwargs={'email': email_ob.to_json()})
+    task_id = task.task_id
+    return {'task_id': task_id}
+
+
+@tasks_router.get(
+    "/taskstatus",
+    description="endpoint to retrieve task status"
+)
+async def taskstatus(task_id):
+    task = AsyncResult(task_id)
+    if isinstance(task.result, Exception):
+        task_result = str(task.result)
+    else:
+        task_result = task.result
+
+    response = {
+        'state': task.state,
+        'result': task_result,
+    }
+
+    return response

fastapi_2fa/api/endpoints/api_v1/router.py renamed to fastapi_2fa/api/endpoints/router.py

@@ -1,6 +1,7 @@
 from fastapi import APIRouter
 
-from fastapi_2fa.api.endpoints.api_v1 import auth, two_factor_auth, users
+from fastapi_2fa.api.endpoints.api_v1 import (auth, tasks, two_factor_auth,
+                                              users)
 
 router = APIRouter()
 
@@ -12,3 +13,6 @@
 
 # tfa
 router.include_router(two_factor_auth.tfa_router, prefix="/tfa", tags=["tfa"])
+
+# tasks
+router.include_router(tasks.tasks_router, prefix="/tasks", tags=["tasks"])

fastapi_2fa/core/config.py

@@ -15,13 +15,21 @@ class BaseConfig(BaseSettings):
 
     FAKE_EMAIL_SENDER: EmailStr = os.environ.get("FAKE_EMAIL_SENDER")
 
+    # # JWT
     JWT_SECRET_KEY: str = os.environ.get("JWT_SECRET_KEY")
     JWT_SECRET_KEY_REFRESH: str = os.environ.get("JWT_SECRET_KEY_REFRESH")
     PRE_TFA_SECRET_KEY: str = os.environ.get("PRE_TFA_SECRET_KEY")
     ALGORITHM: str = os.environ.get("ALGORITHM")
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = os.environ.get("ACCESS_TOKEN_EXPIRE_MINUTES", 15)
+    REFRESH_TOKEN_EXPIRE_MINUTES: int = os.environ.get(
+        "REFRESH_TOKEN_EXPIRE_MINUTES", 60 * 24
+    )  # 24 h
 
+    # # 2 FACTOR AUTHENTICATION
     FERNET_KEY_TFA_TOKEN: str = os.environ.get("FERNET_KEY_TFA_TOKEN")
-
+    PRE_TFA_TOKEN_EXPIRE_MINUTES: int = os.environ.get(
+        "PRE_TFA_TOKEN_EXPIRE_MINUTES", 2
+    )
     TFA_BACKUP_TOKENS_NR: int = os.environ.get("TFA_BACKUP_TOKENS_NR")
     TFA_TOKEN_LENGTH: int = os.environ.get("TFA_TOKEN_LENGTH")
     TOTP_ISSUER_NAME: str = os.environ.get("TOTP_ISSUER_NAME")
@@ -34,18 +42,12 @@ class BaseConfig(BaseSettings):
         le=10
     )
 
-    ACCESS_TOKEN_EXPIRE_MINUTES: int = os.environ.get("ACCESS_TOKEN_EXPIRE_MINUTES", 15)
-    REFRESH_TOKEN_EXPIRE_MINUTES: int = os.environ.get(
-        "REFRESH_TOKEN_EXPIRE_MINUTES", 60 * 24
-    )  # 24 h
-    PRE_TFA_TOKEN_EXPIRE_MINUTES: int = os.environ.get(
-        "PRE_TFA_TOKEN_EXPIRE_MINUTES", 2
-    )
-
+    # # CORS
     BACKEND_CORS_ORIGINS: AnyHttpUrl | list[AnyHttpUrl] = os.environ.get(
         "BACKEND_CORS_ORIGINS", "http://localhost:5555"
     )
 
+    # # DB
     SQLALCHEMY_DATABASE_URI: str = os.environ.get("SQLALCHEMY_DATABASE_URI")
 
     @validator("SQLALCHEMY_DATABASE_URI", pre=True)
@@ -62,6 +64,16 @@ def assemble_db_connection(cls, v: Optional[str], values: Dict[str, Any]) -> Any
             path=f"/{values.get('POSTGRES_DB') or ''}",
         )
 
+    # # CELERY
+    CELERY_BROKER_URL: str = os.environ.get("CELERY_BROKER_URL", "redis://redis:6379/0")
+    result_backend: str = os.environ.get("result_backend", "redis://redis:6379/0")
+    CELERY_WORKER_PREFETCH_MULTIPLIER = 1
+    # Celery ACK the queue only when task is completed
+    CELERY_TASK_ACKS_LATE = True
+    imports = ("fastapi_2fa.tasks.tasks",)
+    task_serializer = "pickle"
+    accept_content = ('pickle', 'json',)
+
     class Config:
         case_sensitive = True
 

fastapi_2fa/core/utils.py

@@ -1,11 +1,13 @@
-from dataclasses import dataclass
+import json
+from dataclasses import asdict, dataclass
 
+from celery.result import AsyncResult
 from pydantic import EmailStr
 
 from fastapi_2fa.core.config import settings
 from fastapi_2fa.models.device import Device
 from fastapi_2fa.models.users import User
-from fastapi_2fa.tasks.tasks import send_email
+from fastapi_2fa.tasks.tasks import send_email_task
 
 
 @dataclass(slots=True, frozen=True)
@@ -14,6 +16,12 @@ class Email:
     from_: EmailStr
     text_: str
 
+    def to_json(self):
+        """
+        get the json formated string
+        """
+        return json.dumps(asdict(self))
+
     def __repr__(self) -> str:
         return (
             f"Email (to: {self.to_}), "
@@ -22,19 +30,21 @@ def __repr__(self) -> str:
         )
 
 
-def send_mail_backup_tokens(user: User, device: Device):
+def send_mail_backup_tokens(user: User, device: Device) -> AsyncResult:
     email_ob = Email(
         to_=[user.email],
         from_=settings.FAKE_EMAIL_SENDER,
         text_=f"Backup tokens : {device.backup_tokens}"
     )
-    send_email(email=email_ob)
+    task = send_email_task.apply_async(kwargs={'email': email_ob.to_json()})
+    return task
 
 
-def send_mail_totp_token(user: User, token: str):
+def send_mail_totp_token(user: User, token: str) -> AsyncResult:
     email_ob = Email(
         to_=[user.email],
         from_=settings.FAKE_EMAIL_SENDER,
         text_=f"Access TOTP token : {token}"
     )
-    send_email(email=email_ob)
+    task = send_email_task.apply_async(kwargs={'email': email_ob.to_json()})
+    return task

fastapi_2fa/crud/users.py

@@ -54,5 +54,4 @@ async def get_by_email(db: Session, email: EmailStr) -> User | None:
         return model_instance
 
 
-
 user_crud = UserCrud(model=User)

fastapi_2fa/main.py

@@ -2,7 +2,7 @@
 from fastapi import FastAPI
 from starlette.middleware.cors import CORSMiddleware
 
-from fastapi_2fa.api.endpoints.api_v1.router import router
+from fastapi_2fa.api.endpoints.router import router
 from fastapi_2fa.core.config import settings
 
 app = FastAPI(

fastapi_2fa/tasks/celery_conf.py

@@ -0,0 +1,5 @@
+from celery import Celery
+
+from fastapi_2fa.core.config import settings
+
+celery = Celery('my_celery', config_source=settings, namespace='CELERY')