Add functionality for email confirmation and password reset

README.md

@@ -40,6 +40,7 @@ Backlog — сервис «списка хотелок» (backlog) для фил
 - Alembic (миграции)
 - PostgreSQL
 - taskiq (фоновые задачи)
+- RabbitMQ
 - Jinja2
 
 ### Frontend
@@ -139,6 +140,12 @@ curl -X POST "http://localhost:8000/api/movies" \
 - Защищённые эндпоинты требуют заголовок:
   `Authorization: Bearer <access_token>`
 
+
+## Брокер сообщений
+- Запуск:
+```bash
+taskiq worker backlog_app.taskiq_broker:broker --fs-discover -tp "**/tasks" --no-configure-logging
+```
 ## Тестирование
 
 - Запуск тестов:

backend/backlog_app/api/view/auth_view.py

@@ -14,7 +14,7 @@
 router.include_router(
     fastapi_users.get_auth_router(
         authentication_backend,
-        # requires_verification=True,  # It is not necessary to include the parameter in this application.
+        requires_verification=True,  # It is not necessary to include the parameter in this application.
     )
 )
 

backend/backlog_app/config.py

@@ -2,7 +2,7 @@
 from pathlib import Path
 from typing import Literal
 
-from pydantic import BaseModel, field_validator
+from pydantic import AmqpDsn, BaseModel, field_validator
 from pydantic_settings import (
     BaseSettings,
     PydanticBaseSettingsSource,
@@ -17,6 +17,9 @@ class LoggingConfig(BaseModel):
     log_format: str = (
         "[-] %(asctime)s [%(levelname)s] %(module)s-%(lineno)d - %(message)s"
     )
+    worker_log_format: str = (
+        "[-] %(asctime)s [%(levelname)s] [%(processName)s] %(module)s-%(lineno)d - %(message)s"
+    )
     log_level_name: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "WARNING"
     log_date_format: str = "%Y-%m-%d %H:%M:%S"
 
@@ -52,6 +55,10 @@ class DataBase(BaseModel):
     connection: DataBaseConnection
 
 
+class TaskiqConfig(BaseModel):
+    url: AmqpDsn = "amqp://guest:guest@localhost:5672//"
+
+
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(
         case_sensitive=False,
@@ -99,6 +106,7 @@ def settings_customise_sources(
         )
 
     db: DataBase
+    taskiq: TaskiqConfig = TaskiqConfig()
     logging: LoggingConfig = LoggingConfig()
     access_token_db: AccessToken
     superuser: SuperUser

backend/backlog_app/servicies/authentification/user_manager.py

@@ -6,10 +6,11 @@
 
 from backlog_app.config import settings
 from backlog_app.models import User
-from backlog_app.tasks.email_task import send_email_confirmed, send_verification_email
+from backlog_app.servicies.mailing import format_seconds_for_email
+from backlog_app.tasks import email_task
 
 if TYPE_CHECKING:
-    from fastapi import Request
+    from fastapi import Request, Response
 
 logger = logging.getLogger(__name__)
 
@@ -19,36 +20,67 @@ class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
     verification_token_secret = settings.access_token_db.verification_token_secret
 
     async def on_after_register(self, user: User, request: Optional["Request"] = None):
-        logger.warning("User %s has registered.", user.id)
+        logger.info("User <%s> has registered.", user.id)
 
-    async def on_after_forgot_password(
+    async def on_after_request_verify(
         self, user: User, token: str, request: Optional["Request"] = None
     ):
-        logger.warning(
-            "User %s has forgot their password. Reset token: $s", user.id, token
+        logger.debug(
+            "Verification requested for user <%s>. Verification token: %s",
+            user.id,
+            token,
         )
 
-    async def on_after_request_verify(
+        origin = request.headers.get("origin") or settings.FRONTEND_URL
+        logger.debug("origin url: %s", origin)
+
+        verification_link = f"{origin}/verify?token={token}"
+        await email_task.send_verification_email.kiq(
+            user_email=user.email,
+            verification_link=verification_link,
+        )
+
+    async def on_after_verify(self, user: User, request: Optional["Request"] = None):
+        logger.warning("User <%s> has been verified", user.id)
+
+        await email_task.send_email_confirmed.kiq(
+            user_email=user.email,
+        )
+
+    async def on_after_login(
+        self,
+        user: User,
+        request: Optional["Request"] = None,
+        response: Optional["Response"] | None = None,
+    ) -> None:
+        logger.warning("User <%s> has logged in", user.id)
+
+    async def on_after_forgot_password(
         self, user: User, token: str, request: Optional["Request"] = None
     ):
-        logger.warning(
-            "Verification requested for user %s. Verification token: %s", user.id, token
+        logger.debug(
+            "User <%s> has forgot their password. Reset token: %s, lifetime: %s",
+            user.id,
+            token,
+            self.reset_password_token_lifetime_seconds,
         )
-
-        verification_link = (
-            "http://127.0.0.1:8000/docs#/Auth/verify_verify_api_auth_verify_post"
+        origin = request.headers.get("origin") or settings.FRONTEND_URL
+        reset_link = f"{origin}/reset-password?token={token}"
+        token_lifetime = format_seconds_for_email(
+            self.reset_password_token_lifetime_seconds
         )
-        await send_verification_email.kiq(
-            user_id=str(user.id),
+
+        await email_task.send_email_forgot_password.kiq(
             user_email=user.email,
-            verification_token=token,
-            verification_link=verification_link,
+            reset_link=reset_link,
+            token_lifetime=token_lifetime,
         )
 
-    async def on_after_verify(self, user: User, request: Optional["Request"] = None):
-        logger.warning("User %s has been verified", user.id)
+    async def on_after_reset_password(
+        self, user: User, request: Optional["Request"] = None
+    ) -> None:
+        logger.warning("User <%s> successfully changed their password", user.id)
 
-        await send_email_confirmed.kiq(
-            user_id=str(user.id),
+        await email_task.send_email_forgot_password_confirmed.kiq(
             user_email=user.email,
         )

backend/backlog_app/servicies/mailing/__init__.py

@@ -0,0 +1,2 @@
+from .email_formater import format_seconds_for_email
+from .email_sender import send_email

backend/backlog_app/servicies/mailing/email_formater.py

@@ -0,0 +1,18 @@
+def format_seconds_for_email(seconds: int) -> str:
+    """
+    Converts a number of seconds into a human-readable string for emails.
+
+    Examples:
+        3600  -> "1 hour"
+        5400  -> "1 hour 30 min"
+        2700  -> "45 min"
+    """
+    hours = seconds // 3600
+    minutes = (seconds % 3600) // 60
+
+    if hours > 0 and minutes == 0:
+        return f"{hours} hour{'s' if hours > 1 else ''}"
+    elif hours > 0 and minutes > 0:
+        return f"{hours} hour{'s' if hours > 1 else ''} {minutes} min"
+    else:
+        return f"{minutes} min"

backend/backlog_app/taskiq_broker.py

@@ -1,3 +1,24 @@
-from taskiq import InMemoryBroker
+import logging
 
-broker = InMemoryBroker()
+import taskiq_fastapi
+from taskiq import TaskiqEvents, TaskiqState
+from taskiq_aio_pika import AioPikaBroker
+
+from backlog_app.config import settings
+
+logger = logging.getLogger(__name__)
+
+broker = AioPikaBroker(url=settings.taskiq.url)
+
+taskiq_fastapi.init(broker, "backlog_app.main:app")
+
+
+@broker.on_event(TaskiqEvents.WORKER_STARTUP)
+async def on_worker_startup(state: TaskiqState) -> None:
+    logging.basicConfig(
+        format=settings.logging.worker_log_format,
+        level=settings.logging.log_level,
+        datefmt=settings.logging.log_date_format,
+    )
+
+    logger.info("Worker startup complete, got state: %s", state)

backend/backlog_app/tasks/__init__.py

@@ -0,0 +1,6 @@
+from .email_task import (
+    send_email_confirmed,
+    send_email_forgot_password,
+    send_email_forgot_password_confirmed,
+    send_verification_email,
+)

backend/backlog_app/tasks/email_task.py

@@ -1,35 +1,30 @@
 from textwrap import dedent
 
 from backlog_app.jinja2_templates import templates
-from backlog_app.servicies.mailing import send_email
+from backlog_app.servicies.mailing.email_sender import send_email
 from backlog_app.taskiq_broker import broker
 
 
 @broker.task
 async def send_verification_email(
-    user_id: str,
     user_email: str,
-    verification_token: str,
     verification_link: str,
-):
-    subject = "Confirm your email for site.com"
+) -> None:
+    subject = "Confirm your email"
 
     plain_content = dedent(f"""\
         Dear {user_email},
         Please verify your email for site.com at {verification_link}.
 
-        Use this token to verify your email: {verification_token}
-
         Your site admin,
         2025
     """)
     template = templates.get_template("email-verify/verification-request.html")
     context = {
-        "user_id": user_id,
+        "user_email": user_email,
         "verification_link": verification_link,
-        "verification_token": verification_token,
     }
-    html_content = template.render(context=context)
+    html_content = template.render(context)
 
     await send_email(
         recipient=user_email,
@@ -41,7 +36,6 @@ async def send_verification_email(
 
 @broker.task
 async def send_email_confirmed(
-    user_id: str,
     user_email: str,
 ):
     subject = "Email Confirmed"
@@ -53,10 +47,60 @@ async def send_email_confirmed(
         2025""")
     template = templates.get_template("email-verify/email-verified.html")
     context = {
-        "user_id": user_id,
         "user_email": user_email,
     }
-    html_content = template.render(context=context)
+    html_content = template.render(context)
+
+    await send_email(
+        recipient=user_email,
+        subject=subject,
+        plain_content=plain_content,
+        html_content=html_content,
+    )
+
+
+@broker.task
+async def send_email_forgot_password(
+    user_email: str, reset_link: str, token_lifetime: str
+):
+    subject = "Request for change password"
+    plain_content = dedent(f"""\
+        Dear {user_email},
+        We get your request for change password.
+        Link for change:  {reset_link}
+        Your site admin,
+            2025""")
+    template = templates.get_template("email-forgot/password-reset-request.html")
+    context = {
+        "user_email": user_email,
+        "reset_link": reset_link,
+        "expires_in": token_lifetime,
+    }
+    html_content = template.render(context)
+
+    await send_email(
+        recipient=user_email,
+        subject=subject,
+        plain_content=plain_content,
+        html_content=html_content,
+    )
+
+
+@broker.task
+async def send_email_forgot_password_confirmed(
+    user_email: str,
+):
+    subject = "Password reset request confirmed"
+    plain_content = dedent(f"""\
+            Dear {user_email},
+            Your password reset request has been confirmed.
+            Your site admin,
+                2025""")
+    template = templates.get_template("email-forgot/password-reset-confirmed.html")
+    context = {
+        "user_email": user_email,
+    }
+    html_content = template.render(context)
 
     await send_email(
         recipient=user_email,

backend/backlog_app/templates/base.html

@@ -10,13 +10,5 @@
     {% block main %}
     {% endblock %}
 </main>
-<footer>
-    {% block footer %}
-    Your site admin,
-    <br>
-    &copy; 2025.
-    {% endblock %}
-</footer>
-
 </body>
 </html>