kytos.core.auth.UserController.get_user() - Code Metrics - Inspection of "feat: replaced `werkzeug/flask` with `uvicorn/star..." - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#375)

by Vinicius

created 2023-05-01 14:28 UTC

kytos.core.auth.UserController.get_user() A

↳ Parent: kytos.core.auth

Complexity

Conditions

Size

Total Lines	12
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	10
nop	2
dl	0
loc	12
rs	9.9
c	0
b	0
f	0

"""Module with main classes related to Authentication."""

# pylint: disable=invalid-name
import base64
import binascii
import datetime
import getpass
import logging
import os
import uuid
from functools import wraps
from http import HTTPStatus

import jwt
import pymongo
from pydantic import ValidationError
from pymongo.collection import ReturnDocument
from pymongo.errors import AutoReconnect, DuplicateKeyError
from pymongo.results import InsertOneResult
from tenacity import retry_if_exception_type, stop_after_attempt, wait_random

from kytos.core.config import KytosConfig
from kytos.core.db import Mongo
from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
                                 content_type_json_or_415, get_json_or_400)
from kytos.core.retry import before_sleep, for_all_methods, retries
from kytos.core.user import HashSubDoc, UserDoc, UserDocUpdate

__all__ = ['authenticated']

LOG = logging.getLogger(__name__)


def authenticated(func):
    """Handle tokens from requests."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        """Verify the requires of token."""
        try:
            request: Request = None
            for arg in args:
                if isinstance(arg, Request):
                    request = arg
                    break
            else:
                raise ValueError("Request arg not found in the decorated func")
            content = request.headers.get("Authorization")
            if content is None:
                raise ValueError("The attribute 'content' has an invalid "
                                 "value 'None'.")
            token = content.split("Bearer ")[1]
            jwt.decode(token, key=Auth.get_jwt_secret(),
                       algorithms=[Auth.encode_algorithm])
        except (
            ValueError,
            IndexError,
            jwt.exceptions.ExpiredSignatureError,
            jwt.exceptions.DecodeError,
        ) as exc:
            msg = f"Token not sent or expired: {exc}"
            return JSONResponse({"error": msg},
                                status_code=HTTPStatus.UNAUTHORIZED.value)
        return func(*args, **kwargs)

    return wrapper


@for_all_methods(
    retries,
    stop=stop_after_attempt(
        int(os.environ.get("MONGO_AUTO_RETRY_STOP_AFTER_ATTEMPT", 3))
    ),
    wait=wait_random(
        min=int(os.environ.get("MONGO_AUTO_RETRY_WAIT_RANDOM_MIN", 0.1)),
        max=int(os.environ.get("MONGO_AUTO_RETRY_WAIT_RANDOM_MAX", 1)),
    ),
    before_sleep=before_sleep,
    retry=retry_if_exception_type((AutoReconnect,)),
)
class UserController:
    """UserController"""

    # pylint: disable=unnecessary-lambda
    def __init__(self, get_mongo=lambda: Mongo()):

        self.mongo = get_mongo()
        self.db_client = self.mongo.client
        self.db = self.db_client[self.mongo.db_name]

    def bootstrap_indexes(self):
        """Bootstrap all users related indexes."""
        index_tuples = [
            ("users", [("username", pymongo.ASCENDING)], {"unique": True}),
        ]
        for collection, keys, kwargs in index_tuples:
            if self.mongo.bootstrap_index(collection, keys, **kwargs):
                LOG.info(
                    f"Created DB index {keys}, collection: {collection}"
                )

    def create_user(self, user_data: dict) -> InsertOneResult:
        """Create user to database"""
        try:
            utc_now = datetime.datetime.utcnow()
            result = self.db.users.insert_one(UserDoc(**{
                "_id": str(uuid.uuid4()),
                "username": user_data.get('username'),
                "hash": HashSubDoc(),
                "password": user_data.get('password'),
                "email": user_data.get('email'),
                "inserted_at": utc_now,
                "updated_at": utc_now,
            }).dict())
        except DuplicateKeyError as err:
            raise err
        except ValidationError as err:
            raise err
        return result

    def delete_user(self, username: str) -> dict:
        """Delete user from database"""
        utc_now = datetime.datetime.utcnow()
        result = self.db.users.find_one_and_update(
            {"username": username},
            {"$set": {
                "state": "inactive",
                "deleted_at": utc_now,
            }},
            return_document=ReturnDocument.AFTER,
        )
        return result

    def update_user(self, username: str, data: dict) -> dict:
        """Update user from database"""
        utc_now = datetime.datetime.utcnow()
        if "password" in data:
            data["hash"] = HashSubDoc()
        try:
            result = self.db.users.find_one_and_update(
                {"username": username},
                {
                    "$set": UserDocUpdate(**{
                        **data,
                        **{"updated_at": utc_now}
                    }).dict(exclude_none=True)
                },
                return_document=ReturnDocument.AFTER
            )
        except ValidationError as err:
            raise err
        except DuplicateKeyError as err:
            raise err
        return result

    def get_user(self, username: str) -> dict:
        """Return a user information from database"""
        data = self.db.users.aggregate([
            {"$match": {"username": username}},
            {"$project": UserDoc.projection()},
            {"$limit": 1}
        ])
        try:
            user, *_ = list(value for value in data)
            return user
        except ValueError:
            return {}

    def get_user_nopw(self, username: str) -> dict:
        """Return a user information from database without password"""
        data = self.db.users.aggregate([
            {"$match": {"username": username}},
            {"$project": UserDoc.projection_nopw()},
            {"$limit": 1}
        ])
        try:
            user, *_ = list(value for value in data)
            return user
        except ValueError:
            return {}

    def get_users(self) -> dict:
        """Return all the users"""
        data = self.db.users.aggregate([
            {"$project": UserDoc.projection_nopw()}
        ])
        return {'users': list(value for value in data)}


class Auth:
    """Module used to provide Kytos authentication routes."""

    encode_algorithm = "HS256"

    def __init__(self, controller):
        """Init method of Auth class takes the parameters below.

        Args:
            controller(kytos.core.controller): A Controller instance.

        """
        self.user_controller = self.get_user_controller()
        self.user_controller.bootstrap_indexes()
        self.controller = controller
        self.token_expiration_minutes = self.get_token_expiration()
        if self.controller.options.create_superuser is True:
            self._create_superuser()

    @staticmethod
    def get_user_controller():
        """Get UserController"""
        return UserController()

    @staticmethod
    def get_token_expiration():
        """Return token expiration time in minutes defined in kytos conf."""
        options = KytosConfig().options['daemon']
        return options.token_expiration_minutes

    @classmethod
    def get_jwt_secret(cls):
        """Return JWT secret defined in kytos conf."""
        options = KytosConfig().options['daemon']
        return options.jwt_secret

    @classmethod
    def _generate_token(cls, username, time_exp):
        """Generate a jwt token."""
        return jwt.encode(
            {
                'username': username,
                'iss': "Kytos NApps Server",
                'exp': time_exp,
            },
            Auth.get_jwt_secret(),
            algorithm=cls.encode_algorithm,
        )

    def _create_superuser(self):
        """Create a superuser using MongoDB."""
        def get_username():
            return input("Username: ")

        def get_email():
            return input("Email: ")

        username = get_username()
        email = get_email()
        while True:
            password = getpass.getpass()
            re_password = getpass.getpass('Retype password: ')
            if password == re_password:
                break
        user = {
            "username": username,
            "email": email,
            "password": password,
        }
        try:
            self.user_controller.create_user(user)
        except ValidationError as err:
            msg = self.error_msg(err.errors())
            return f"Error: {msg}"
        except DuplicateKeyError:
            return f"Error: {username} already exist."

        return "User successfully created"

    def register_core_auth_services(self):
        """
        Register /kytos/core/ services over authentication.

        It registers create, authenticate, list all, list specific, delete and
        update users.
        """
        self.controller.api_server.register_core_endpoint(
            "auth/login/", self._authenticate_user
        )
        self.controller.api_server.register_core_endpoint(
            "auth/users/", self._list_users,
        )
        self.controller.api_server.register_core_endpoint(
            "auth/users/{username}", self._list_user
        )
        self.controller.api_server.register_core_endpoint(
            "auth/users/", self._create_user, methods=["POST"]
        )
        self.controller.api_server.register_core_endpoint(
            "auth/users/{username}", self._delete_user, methods=["DELETE"]
        )
        self.controller.api_server.register_core_endpoint(
            "auth/users/{username}", self._update_user, methods=["PATCH"]
        )

    def _authenticate_user(self, request: Request) -> JSONResponse:
        """Authenticate a user using MongoDB."""
        if "Authorization" not in request.headers:
            raise HTTPException(400, detail="Authorization header missing")
        try:
            auth = request.headers["Authorization"]
            _scheme, credentials = auth.split()
            decoded = base64.b64decode(credentials).decode("ascii")
            username, _, password = decoded.partition(":")
            password = password.encode()
        except (ValueError, TypeError, binascii.Error) as err:
            msg = "Credentials were not correctly set."
            raise HTTPException(400, detail=msg) from err
        user = self._find_user(username)
        if user["state"] != 'active':
            raise HTTPException(401, detail='This user is not active')
        password_hashed = UserDoc.hashing(password, user["hash"])
        if user["password"] != password_hashed:
            raise HTTPException(401, detail="Incorrect password")
        time_exp = datetime.datetime.utcnow() + datetime.timedelta(
            minutes=self.token_expiration_minutes
        )
        token = self._generate_token(username, time_exp)
        return JSONResponse({"token": token})

    def _find_user(self, username):
        """Find a specific user using MongoDB."""
        user = self.user_controller.get_user(username)
        if not user:
            raise HTTPException(404, detail=f"User {username} not found")
        return user

    @authenticated
    def _list_user(self, request: Request) -> JSONResponse:
        """List a specific user using MongoDB."""
        username = request.path_params["username"]
        user = self.user_controller.get_user_nopw(username)
        if not user:
            raise HTTPException(404, detail=f"User {username} not found")
        return JSONResponse(user)

    @authenticated
    def _list_users(self, _request: Request) -> JSONResponse:
        """List all users using MongoDB."""
        users_list = self.user_controller.get_users()
        return JSONResponse(users_list)

    def _get_request_body(self, request: Request) -> dict:
        """Get JSON from request"""
        content_type_json_or_415(request)
        body = get_json_or_400(request, self.controller.loop)
        if not isinstance(body, dict):
            raise HTTPException(400, "Invalid payload type: {body}")
        return body

    @staticmethod
    def error_msg(error_list: list) -> str:
        """Return a more request friendly error message from ValidationError"""
        msg = ""
        for err in error_list:
            for value in err['loc']:
                msg += value + ", "
            msg = msg[:-2]
            msg += ": " + err["msg"] + "; "
        return msg[:-2]

    @authenticated
    def _create_user(self, request: Request) -> JSONResponse:
        """Save a user using MongoDB."""
        try:
            user_data = self._get_request_body(request)
            self.user_controller.create_user(user_data)
        except ValidationError as err:
            msg = self.error_msg(err.errors())
            raise HTTPException(400, msg) from err
        except DuplicateKeyError as err:
            msg = user_data.get("username") + " already exists."
            raise HTTPException(409, detail=msg) from err
        return JSONResponse("User successfully created", status_code=201)

    @authenticated
    def _delete_user(self, request: Request) -> JSONResponse:
        """Delete a user using MongoDB."""
        username = request.path_params["username"]
        if not self.user_controller.delete_user(username):
            raise HTTPException(404, detail=f"User {username} not found")
        return JSONResponse(f"User {username} deleted succesfully")

    @authenticated
    def _update_user(self, request: Request) -> JSONResponse:
        """Update user data using MongoDB."""
        body = self._get_request_body(request)
        username = request.path_params["username"]
        allowed = ["username", "email", "password"]
        data = {}
        for key, value in body.items():
            if key in allowed:
                data[key] = value
        try:
            updated = self.user_controller.update_user(username, data)
        except ValidationError as err:
            msg = self.error_msg(err.errors())
            raise HTTPException(400, detail=msg) from err
        except DuplicateKeyError as err:
            msg = username + " already exists."
            raise HTTPException(409, detail=msg) from err
        if not updated:
            raise HTTPException(404, detail=f"User {username} not found")
        return JSONResponse("User successfully updated")


1			"""Module with main classes related to Authentication."""
2
3			# pylint: disable=invalid-name
4			import base64
5			import binascii
6			import datetime
7			import getpass
8			import logging
9			import os
10			import uuid
11			from functools import wraps
12			from http import HTTPStatus
13
14			import jwt
15			import pymongo
16			from pydantic import ValidationError
17			from pymongo.collection import ReturnDocument
18			from pymongo.errors import AutoReconnect, DuplicateKeyError
19			from pymongo.results import InsertOneResult
20			from tenacity import retry_if_exception_type, stop_after_attempt, wait_random
21
22			from kytos.core.config import KytosConfig
23			from kytos.core.db import Mongo
24			from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
25			content_type_json_or_415, get_json_or_400)
26			from kytos.core.retry import before_sleep, for_all_methods, retries
27			from kytos.core.user import HashSubDoc, UserDoc, UserDocUpdate
28
29			__all__ = ['authenticated']
30
31			LOG = logging.getLogger(__name__)
32
33
34			def authenticated(func):
35			"""Handle tokens from requests."""
36			@wraps(func)
37			def wrapper(args, *kwargs):
38			"""Verify the requires of token."""
39			try:
40			request: Request = None
41			for arg in args:
42			if isinstance(arg, Request):
43			request = arg
44			break
45			else:
46			raise ValueError("Request arg not found in the decorated func")
47			content = request.headers.get("Authorization")
48			if content is None:
49			raise ValueError("The attribute 'content' has an invalid "
50			"value 'None'.")
51			token = content.split("Bearer ")[1]
52			jwt.decode(token, key=Auth.get_jwt_secret(),
53			algorithms=[Auth.encode_algorithm])
54			except (
55			ValueError,
56			IndexError,
57			jwt.exceptions.ExpiredSignatureError,
58			jwt.exceptions.DecodeError,
59			) as exc:
60			msg = f"Token not sent or expired: {exc}"
61			return JSONResponse({"error": msg},
62			status_code=HTTPStatus.UNAUTHORIZED.value)
63			return func(args, *kwargs)
64
65			return wrapper
66
67
68			@for_all_methods(
69			retries,
70			stop=stop_after_attempt(
71			int(os.environ.get("MONGO_AUTO_RETRY_STOP_AFTER_ATTEMPT", 3))
72			),
73			wait=wait_random(
74			min=int(os.environ.get("MONGO_AUTO_RETRY_WAIT_RANDOM_MIN", 0.1)),
75			max=int(os.environ.get("MONGO_AUTO_RETRY_WAIT_RANDOM_MAX", 1)),
76			),
77			before_sleep=before_sleep,
78			retry=retry_if_exception_type((AutoReconnect,)),
79			)
80			class UserController:
81			"""UserController"""
82
83			# pylint: disable=unnecessary-lambda
84			def __init__(self, get_mongo=lambda: Mongo()):
			0 ignored issues – show Comprehensibility Best Practice introduced 2022-10-12 23:54 UTC by Report Bug Copy Issue Report The variable `Mongo` does not seem to be defined. Loading history...
85			self.mongo = get_mongo()
86			self.db_client = self.mongo.client
87			self.db = self.db_client[self.mongo.db_name]
88
89			def bootstrap_indexes(self):
90			"""Bootstrap all users related indexes."""
91			index_tuples = [
92			("users", [("username", pymongo.ASCENDING)], {"unique": True}),
93			]
94			for collection, keys, kwargs in index_tuples:
95			if self.mongo.bootstrap_index(collection, keys, **kwargs):
96			LOG.info(
97			f"Created DB index {keys}, collection: {collection}"
98			)
99
100			def create_user(self, user_data: dict) -> InsertOneResult:
101			"""Create user to database"""
102			try:
103			utc_now = datetime.datetime.utcnow()
104			result = self.db.users.insert_one(UserDoc(**{
105			"_id": str(uuid.uuid4()),
106			"username": user_data.get('username'),
107			"hash": HashSubDoc(),
108			"password": user_data.get('password'),
109			"email": user_data.get('email'),
110			"inserted_at": utc_now,
111			"updated_at": utc_now,
112			}).dict())
113			except DuplicateKeyError as err:
114			raise err
115			except ValidationError as err:
116			raise err
117			return result
118
119			def delete_user(self, username: str) -> dict:
120			"""Delete user from database"""
121			utc_now = datetime.datetime.utcnow()
122			result = self.db.users.find_one_and_update(
123			{"username": username},
124			{"$set": {
125			"state": "inactive",
126			"deleted_at": utc_now,
127			}},
128			return_document=ReturnDocument.AFTER,
129			)
130			return result
131
132			def update_user(self, username: str, data: dict) -> dict:
133			"""Update user from database"""
134			utc_now = datetime.datetime.utcnow()
135			if "password" in data:
136			data["hash"] = HashSubDoc()
137			try:
138			result = self.db.users.find_one_and_update(
139			{"username": username},
140			{
141			"$set": UserDocUpdate(**{
142			**data,
143			**{"updated_at": utc_now}
144			}).dict(exclude_none=True)
145			},
146			return_document=ReturnDocument.AFTER
147			)
148			except ValidationError as err:
149			raise err
150			except DuplicateKeyError as err:
151			raise err
152			return result
153
154			def get_user(self, username: str) -> dict:
155			"""Return a user information from database"""
156			data = self.db.users.aggregate([
157			{"$match": {"username": username}},
158			{"$project": UserDoc.projection()},
159			{"$limit": 1}
160			])
161			try:
162			user, *_ = list(value for value in data)
163			return user
164			except ValueError:
165			return {}
166
167			def get_user_nopw(self, username: str) -> dict:
168			"""Return a user information from database without password"""
169			data = self.db.users.aggregate([
170			{"$match": {"username": username}},
171			{"$project": UserDoc.projection_nopw()},
172			{"$limit": 1}
173			])
174			try:
175			user, *_ = list(value for value in data)
176			return user
177			except ValueError:
178			return {}
179
180			def get_users(self) -> dict:
181			"""Return all the users"""
182			data = self.db.users.aggregate([
183			{"$project": UserDoc.projection_nopw()}
184			])
185			return {'users': list(value for value in data)}
186
187
188			class Auth:
189			"""Module used to provide Kytos authentication routes."""
190
191			encode_algorithm = "HS256"
192
193			def __init__(self, controller):
194			"""Init method of Auth class takes the parameters below.
195
196			Args:
197			controller(kytos.core.controller): A Controller instance.
198
199			"""
200			self.user_controller = self.get_user_controller()
201			self.user_controller.bootstrap_indexes()
202			self.controller = controller
203			self.token_expiration_minutes = self.get_token_expiration()
204			if self.controller.options.create_superuser is True:
205			self._create_superuser()
206
207			@staticmethod
208			def get_user_controller():
209			"""Get UserController"""
210			return UserController()
211
212			@staticmethod
213			def get_token_expiration():
214			"""Return token expiration time in minutes defined in kytos conf."""
215			options = KytosConfig().options['daemon']
216			return options.token_expiration_minutes
217
218			@classmethod
219			def get_jwt_secret(cls):
220			"""Return JWT secret defined in kytos conf."""
221			options = KytosConfig().options['daemon']
222			return options.jwt_secret
223
224			@classmethod
225			def _generate_token(cls, username, time_exp):
226			"""Generate a jwt token."""
227			return jwt.encode(
228			{
229			'username': username,
230			'iss': "Kytos NApps Server",
231			'exp': time_exp,
232			},
233			Auth.get_jwt_secret(),
234			algorithm=cls.encode_algorithm,
235			)
236
237			def _create_superuser(self):
238			"""Create a superuser using MongoDB."""
239			def get_username():
240			return input("Username: ")
241
242			def get_email():
243			return input("Email: ")
244
245			username = get_username()
246			email = get_email()
247			while True:
248			password = getpass.getpass()
249			re_password = getpass.getpass('Retype password: ')
250			if password == re_password:
251			break
252			user = {
253			"username": username,
254			"email": email,
255			"password": password,
256			}
257			try:
258			self.user_controller.create_user(user)
259			except ValidationError as err:
260			msg = self.error_msg(err.errors())
261			return f"Error: {msg}"
262			except DuplicateKeyError:
263			return f"Error: {username} already exist."
264
265			return "User successfully created"
266
267			def register_core_auth_services(self):
268			"""
269			Register /kytos/core/ services over authentication.
270
271			It registers create, authenticate, list all, list specific, delete and
272			update users.
273			"""
274			self.controller.api_server.register_core_endpoint(
275			"auth/login/", self._authenticate_user
276			)
277			self.controller.api_server.register_core_endpoint(
278			"auth/users/", self._list_users,
279			)
280			self.controller.api_server.register_core_endpoint(
281			"auth/users/{username}", self._list_user
282			)
283			self.controller.api_server.register_core_endpoint(
284			"auth/users/", self._create_user, methods=["POST"]
285			)
286			self.controller.api_server.register_core_endpoint(
287			"auth/users/{username}", self._delete_user, methods=["DELETE"]
288			)
289			self.controller.api_server.register_core_endpoint(
290			"auth/users/{username}", self._update_user, methods=["PATCH"]
291			)
292
293			def _authenticate_user(self, request: Request) -> JSONResponse:
294			"""Authenticate a user using MongoDB."""
295			if "Authorization" not in request.headers:
296			raise HTTPException(400, detail="Authorization header missing")
297			try:
298			auth = request.headers["Authorization"]
299			_scheme, credentials = auth.split()
300			decoded = base64.b64decode(credentials).decode("ascii")
301			username, _, password = decoded.partition(":")
302			password = password.encode()
303			except (ValueError, TypeError, binascii.Error) as err:
304			msg = "Credentials were not correctly set."
305			raise HTTPException(400, detail=msg) from err
306			user = self._find_user(username)
307			if user["state"] != 'active':
308			raise HTTPException(401, detail='This user is not active')
309			password_hashed = UserDoc.hashing(password, user["hash"])
310			if user["password"] != password_hashed:
311			raise HTTPException(401, detail="Incorrect password")
312			time_exp = datetime.datetime.utcnow() + datetime.timedelta(
313			minutes=self.token_expiration_minutes
314			)
315			token = self._generate_token(username, time_exp)
316			return JSONResponse({"token": token})
317
318			def _find_user(self, username):
319			"""Find a specific user using MongoDB."""
320			user = self.user_controller.get_user(username)
321			if not user:
322			raise HTTPException(404, detail=f"User {username} not found")
323			return user
324
325			@authenticated
326			def _list_user(self, request: Request) -> JSONResponse:
327			"""List a specific user using MongoDB."""
328			username = request.path_params["username"]
329			user = self.user_controller.get_user_nopw(username)
330			if not user:
331			raise HTTPException(404, detail=f"User {username} not found")
332			return JSONResponse(user)
333
334			@authenticated
335			def _list_users(self, _request: Request) -> JSONResponse:
336			"""List all users using MongoDB."""
337			users_list = self.user_controller.get_users()
338			return JSONResponse(users_list)
339
340			def _get_request_body(self, request: Request) -> dict:
341			"""Get JSON from request"""
342			content_type_json_or_415(request)
343			body = get_json_or_400(request, self.controller.loop)
344			if not isinstance(body, dict):
345			raise HTTPException(400, "Invalid payload type: {body}")
346			return body
347
348			@staticmethod
349			def error_msg(error_list: list) -> str:
350			"""Return a more request friendly error message from ValidationError"""
351			msg = ""
352			for err in error_list:
353			for value in err['loc']:
354			msg += value + ", "
355			msg = msg[:-2]
356			msg += ": " + err["msg"] + "; "
357			return msg[:-2]
358
359			@authenticated
360			def _create_user(self, request: Request) -> JSONResponse:
361			"""Save a user using MongoDB."""
362			try:
363			user_data = self._get_request_body(request)
364			self.user_controller.create_user(user_data)
365			except ValidationError as err:
366			msg = self.error_msg(err.errors())
367			raise HTTPException(400, msg) from err
368			except DuplicateKeyError as err:
369			msg = user_data.get("username") + " already exists."
370			raise HTTPException(409, detail=msg) from err
371			return JSONResponse("User successfully created", status_code=201)
372
373			@authenticated
374			def _delete_user(self, request: Request) -> JSONResponse:
375			"""Delete a user using MongoDB."""
376			username = request.path_params["username"]
377			if not self.user_controller.delete_user(username):
378			raise HTTPException(404, detail=f"User {username} not found")
379			return JSONResponse(f"User {username} deleted succesfully")
380
381			@authenticated
382			def _update_user(self, request: Request) -> JSONResponse:
383			"""Update user data using MongoDB."""
384			body = self._get_request_body(request)
385			username = request.path_params["username"]
386			allowed = ["username", "email", "password"]
387			data = {}
388			for key, value in body.items():
389			if key in allowed:
390			data[key] = value
391			try:
392			updated = self.user_controller.update_user(username, data)
393			except ValidationError as err:
394			msg = self.error_msg(err.errors())
395			raise HTTPException(400, detail=msg) from err
396			except DuplicateKeyError as err:
397			msg = username + " already exists."
398			raise HTTPException(409, detail=msg) from err
399			if not updated:
400			raise HTTPException(404, detail=f"User {username} not found")
401			return JSONResponse("User successfully updated")
402

kytos-ng / kytos

Pull Request — master (#375)

kytos.core.auth.UserController.get_user() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like