Added rate limiter for send-email

src/app/main.py

@@ -4,13 +4,18 @@ from fastapi.responses import FileResponse
 
 from pydantic import ValidationError
 
+from slowapi import _rate_limit_exceeded_handler
+from slowapi.errors import RateLimitExceeded
+
 from app.core.config import settings
 from app.views.router import main_router
+from app.utils.rate_limiter import limiter
 
 
 app = FastAPI(
     title=settings.PROJECT_NAME
 )
+app.state.limiter = limiter
 
 # app.mount(
 #     '/static',
@@ -21,6 +26,9 @@ app = FastAPI(
 app.include_router(main_router)
 
 
+app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
+
+
 @app.exception_handler(ValidationError)
 async def validation_exception_handler(request, exc):
     return await request_validation_exception_handler(request, exc)

src/app/utils/custom_functions.py

@@ -1,6 +1,11 @@
 from bs4 import BeautifulSoup
+from fastapi import Request
 
 
 def html2text(html: str) -> str:
     soup = BeautifulSoup(html, 'html.parser')
-    return soup.get_text()
+    return soup.get_text()
+
+
+def get_remote_address(request: Request) -> str:
+    return request.headers.get('host')

src/app/utils/email_utils.py

@@ -13,7 +13,7 @@ from app.core.config import settings
 def send_email_notification(
     subject: str,
     body: str
-) -> partial:
+) -> partial | None:
 
     if settings.EMAIL_RECIPIENTS:
         conf = ConnectionConfig(

src/app/utils/rate_limiter.py

@@ -0,0 +1,4 @@
+from slowapi import Limiter
+from app.utils.custom_functions import get_remote_address
+
+limiter = Limiter(key_func=get_remote_address)

src/app/views/routers/main.py

@@ -22,6 +22,7 @@ from app.schemas.main import SendEmail
 from app.schemas.post import PostDetail
 from app.utils.email_utils import send_email_notification
 from app.models.user import User as UserModel
+from app.utils.rate_limiter import limiter
 from app.views.depends import (
     get_async_db,
     get_current_active_superuser_or_none
@@ -42,6 +43,7 @@ async def home(request: Request):
 
 
 @router.post('/send-email')
+@limiter.limit('2/day')
 async def send_email(
     request: Request,
     background_tasks: BackgroundTasks,
@@ -53,12 +55,15 @@ async def send_email(
             f"phone: {form_data.phone}\n"\
             f"message: {form_data.message}"
 
-    background_tasks.add_task(
-        send_email_notification(
+    email_notification = send_email_notification(
             subject = f"Portfolio Blog (by {form_data.email})",
             body = body
         )
-    )
+
+    if email_notification is not None:
+        background_tasks.add_task(
+            email_notification
+        )
 
     return RedirectResponse(
         str(request.url_for('home')) + '#contact',