51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
from flask_restx import Resource
|
|
from flask import request, current_app
|
|
from flask_jwt_extended import create_access_token
|
|
from app.db.models import UserModel
|
|
from app.docs.auth_models import auth_ns, signup_model, signin_model
|
|
from app.schemas.auth_sigin_schema import SigInRequest
|
|
from app.schemas.auth_sigup_schema import SigUpRequest
|
|
|
|
@auth_ns.route('/signup')
|
|
class SignUp(Resource):
|
|
@auth_ns.expect(signup_model)
|
|
def post(self):
|
|
data = request.get_json()
|
|
|
|
validated = SigUpRequest(**data)
|
|
|
|
roles = data.get("roles", [])
|
|
|
|
user_model = UserModel()
|
|
if user_model.find_by_email(validated.email):
|
|
return {'message': 'User already exists'}, 400
|
|
|
|
result = user_model.create_user(validated.email, validated.password)
|
|
|
|
user_model.update_user(result.inserted_id, {"email": validated.email, "roles": roles})
|
|
|
|
return {'message': 'success'}, 201
|
|
|
|
@auth_ns.route('/login')
|
|
class Login(Resource):
|
|
@auth_ns.expect(signin_model)
|
|
def post(self):
|
|
data = request.get_json()
|
|
|
|
validated = SigInRequest(**data)
|
|
|
|
user_model = UserModel()
|
|
user = user_model.find_by_email(validated.email)
|
|
|
|
if not user or not user_model.verify_password(user['password'], data['password']):
|
|
return {'message': 'Invalid credentials'}, 401
|
|
|
|
roles = user.get("roles", []) if user else []
|
|
|
|
access_token = create_access_token(
|
|
identity=user['email'],
|
|
additional_claims={"roles": roles}
|
|
)
|
|
|
|
return {'access_token': access_token}, 200
|