|
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
|
|
|
|
|
|
class LoginModel(BaseModel):
|
|
email: EmailStr
|
|
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": {}}
|
|
)
|
|
|
|
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}
|
|
)
|
|
|
|
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
|
|
return web.HTTPFound("/dashboard")
|
|
|
|
return aiohttp_jinja2.render_template(
|
|
self.template_name,
|
|
self.request,
|
|
{"error": "Invalid email or password", "request": self.request, "errors": {}},
|
|
)
|
|
|
|
|
|
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": {}}
|
|
)
|
|
|
|
async def post(self):
|
|
try:
|
|
user_data = RegistrationModel(**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 = await self.request.post()
|
|
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": {}},
|
|
)
|
|
|
|
user_service: UserService = self.request.app["user_service"]
|
|
try:
|
|
user_service.create_user(user_data.username, user_data.email, user_data.password)
|
|
except ValueError:
|
|
return aiohttp_jinja2.render_template(
|
|
self.template_name,
|
|
self.request,
|
|
{
|
|
"error": "User with this email already exists",
|
|
"request": self.request,
|
|
"errors": {},
|
|
},
|
|
)
|
|
|
|
return web.HTTPFound("/login")
|
|
|
|
|
|
class LogoutView(web.View):
|
|
async def get(self):
|
|
session = await get_session(self.request)
|
|
session.clear()
|
|
return web.HTTPFound("/")
|