from aiohttp import web
import aiohttp_jinja2
from aiohttp_session import get_session, new_session
from aiohttp_pydantic import PydanticView
from pydantic import BaseModel, EmailStr, Field, ValidationError
from ..services.user_service import UserService
from ..models import RegistrationModel
from ..helpers.email_sender import send_email # Import send_email
class LoginModel(BaseModel):
email: EmailStr
password: str
class ForgotPasswordModel(BaseModel):
email: EmailStr
class ResetPasswordModel(BaseModel):
password: str = Field(min_length=8)
confirm_password: str
class CustomPydanticView(PydanticView):
template_name: str = ""
async def on_validation_error(
self, exception: ValidationError, context: str
):
errors = {
err["loc"][0]: err["msg"] for err in exception.errors()
}
return aiohttp_jinja2.render_template(
self.template_name, self.request, {"errors": errors, "request": self.request}
)
class LoginView(CustomPydanticView):
template_name = "pages/login.html"
async def get(self):
return aiohttp_jinja2.render_template(
self.template_name, self.request, {"request": self.request, "errors": {}, "data": {}}
)
async def post(self):
try:
login_data = LoginModel(**await self.request.post())
except ValidationError as e:
errors = {err["loc"][0]: err["msg"] for err in e.errors()}
return aiohttp_jinja2.render_template(
self.template_name, self.request, {"errors": errors, "request": self.request, "data": dict(await self.request.post())}
)
user_service: UserService = self.request.app["user_service"]
if user_service.authenticate_user(login_data.email, login_data.password):
session = await new_session(self.request)
session["user_email"] = login_data.email
raise web.HTTPFound("/dashboard")
return aiohttp_jinja2.render_template(
self.template_name,
self.request,
{"error": "Invalid email or password", "request": self.request, "errors": {}, "data": dict(await self.request.post())},
)
class RegistrationView(CustomPydanticView):
template_name = "pages/register.html"
async def get(self):
return aiohttp_jinja2.render_template(
self.template_name, self.request, {"request": self.request, "errors": {}, "data": {}}
)
async def post(self):
data = await self.request.post() # Get raw form data once
try:
user_data = RegistrationModel(**data)
except ValidationError as e:
errors = {err["loc"][0]: err["msg"] for err in e.errors()}
return aiohttp_jinja2.render_template(
self.template_name, self.request, {"errors": errors, "request": self.request, "data": dict(data)}
)
if user_data.password != data.get("confirm_password"):
return aiohttp_jinja2.render_template(
self.template_name,
self.request,
{"error": "Passwords do not match", "request": self.request, "errors": {}, "data": dict(data)},
)
user_service: UserService = self.request.app["user_service"]
try:
user_service.create_user(user_data.full_name, user_data.email, user_data.password) # Changed username to full_name
# Render email content
email_context = {
"user_name": user_data.full_name,
"login_url": self.request.url.join(self.request.app.router["login"].url_for()).human_repr(),
"support_url": self.request.url.join(self.request.app.router["support"].url_for()).human_repr(),
}
env = aiohttp_jinja2.get_env(self.request.app)
template = env.get_template("emails/welcome.html")
email_body = template.render(email_context)
# Schedule email sending
await self.request.app["scheduler"].spawn(
send_email(
self.request.app,
user_data.email,
"Welcome to Retoor's Cloud Solutions!",
email_body
)
)
raise web.HTTPFound("/login")
except ValueError:
return aiohttp_jinja2.render_template(
self.template_name,
self.request,
{
"error": "User with this email already exists",
"request": self.request,
"errors": {},
"data": dict(data),
},
)
class ForgotPasswordView(CustomPydanticView):
template_name = "pages/forgot_password.html"
async def get(self):
return aiohttp_jinja2.render_template(
self.template_name, self.request, {"request": self.request, "errors": {}, "data": {}}
)
async def post(self):
data = await self.request.post()
try:
forgot_password_data = ForgotPasswordModel(**data)
except ValidationError as e:
errors = {err["loc"][0]: err["msg"] for err in e.errors()}
return aiohttp_jinja2.render_template(
self.template_name, self.request, {"errors": errors, "request": self.request, "data": dict(data)}
)
user_service: UserService = self.request.app["user_service"]
user = user_service.get_user_by_email(forgot_password_data.email)
if user:
token = user_service.generate_reset_token(forgot_password_data.email)
if token:
reset_link = self.request.url.join(
self.request.app.router["reset_password"].url_for(token=token)
).human_repr()
email_context = {
"user_name": user["full_name"],
"reset_link": reset_link,
}
env = aiohttp_jinja2.get_env(self.request.app)
template = env.get_template("emails/password_reset_request.html")
email_body = template.render(email_context)
await self.request.app["scheduler"].spawn(
send_email(
self.request.app,
forgot_password_data.email,
"Retoor's Cloud Solutions - Password Reset Request",
email_body
)
)
# Always show a success message to prevent email enumeration attacks
return aiohttp_jinja2.render_template(
self.template_name,
self.request,
{"message": "If an account with that email exists, a password reset link has been sent.", "request": self.request, "errors": {}, "data": {}},
)
class ResetPasswordView(CustomPydanticView):
template_name = "pages/reset_password.html"
async def get(self):
token = self.request.match_info.get("token")
if not token:
raise web.HTTPFound("/forgot_password") # Or show an error page
# We don't validate the token here, only when the form is submitted
return aiohttp_jinja2.render_template(
self.template_name, self.request, {"request": self.request, "errors": {}, "token": token}
)
async def post(self):
token = self.request.match_info.get("token")
if not token:
raise web.HTTPFound("/forgot_password") # Or show an error page
data = await self.request.post()
try:
reset_password_data = ResetPasswordModel(**data)
except ValidationError as e:
errors = {err["loc"][0]: err["msg"] for err in e.errors()}
return aiohttp_jinja2.render_template(
self.template_name, self.request, {"errors": errors, "request": self.request, "token": token}
)
if reset_password_data.password != reset_password_data.confirm_password:
return aiohttp_jinja2.render_template(
self.template_name,
self.request,
{"error": "Passwords do not match", "request": self.request, "errors": {}, "token": token},
)
user_service: UserService = self.request.app["user_service"]
user = user_service.get_user_by_reset_token(token) # Corrected method call
if not user:
return aiohttp_jinja2.render_template(
self.template_name,
self.request,
{"error": "Invalid or expired password reset link.", "request": self.request, "errors": {}, "token": token},
)
if user_service.reset_password(user["email"], token, reset_password_data.password):
# Send password changed confirmation email
email_context = {
"user_name": user["full_name"],
"login_url": self.request.url.join(self.request.app.router["login"].url_for()).human_repr(),
}
env = aiohttp_jinja2.get_env(self.request.app)
template = env.get_template("emails/password_changed_confirmation.html")
email_body = template.render(email_context)
await self.request.app["scheduler"].spawn(
send_email(
self.request.app,
user["email"],
"Retoor's Cloud Solutions - Your Password Has Been Changed",
email_body
)
)
raise web.HTTPFound("/login?message=password_reset_success")
else:
return aiohttp_jinja2.render_template(
self.template_name,
self.request,
{"error": "Invalid or expired password reset link.", "request": self.request, "errors": {}, "token": token},
)
class LogoutView(web.View):
async def get(self):
session = await get_session(self.request)
session.clear()
raise web.HTTPFound("/")
Reference in New Issue View Git Blame Copy Permalink