tracim_backend.lib.utils.request.TracimRequest._cleanup() - Code Metrics - Inspection of "Add TODO" - tracim/tracim_v2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( c8e157...770c39 )

by Bastien

created 2018-09-12 08:13 UTC

TracimRequest._cleanup() A

↳ Parent: tracim_backend.lib.utils.request

Complexity

Conditions

Size

Total Lines	12
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# -*- coding: utf-8 -*-
from pyramid.request import Request
from pyramid.response import Response
from sqlalchemy.orm.exc import NoResultFound

from tracim_backend.exceptions import NotAuthenticated
from tracim_backend.exceptions import UserNotActive
from tracim_backend.exceptions import ContentNotFound
from tracim_backend.exceptions import InvalidUserId
from tracim_backend.exceptions import InvalidWorkspaceId
from tracim_backend.exceptions import InvalidContentId
from tracim_backend.exceptions import InvalidCommentId
from tracim_backend.exceptions import ContentNotFoundInTracimRequest
from tracim_backend.exceptions import WorkspaceNotFoundInTracimRequest
from tracim_backend.exceptions import UserNotFoundInTracimRequest
from tracim_backend.exceptions import UserDoesNotExist
from tracim_backend.exceptions import WorkspaceNotFound
from tracim_backend.exceptions import ImmutableAttribute
from tracim_backend.app_models.contents import CONTENT_TYPES
from tracim_backend.lib.core.content import ContentApi
from tracim_backend.lib.core.user import UserApi
from tracim_backend.lib.core.workspace import WorkspaceApi
from tracim_backend.lib.utils.authorization import JSONDecodeError

from tracim_backend.models import User
from tracim_backend.models.data import Workspace
from tracim_backend.models.data import Content


class TracimRequest(Request):
    """
    Request with tracim specific params/methods
    """
    def __init__(
            self,
            environ,
            charset=None,
            unicode_errors=None,
            decode_param_names=None,
            **kw
    ):
        super().__init__(
            environ,
            charset,
            unicode_errors,
            decode_param_names,
            **kw
        )
        # Current comment, found in request path
        self._current_comment = None  # type: Content

        # Current content, found in request path
        self._current_content = None  # type: Content

        # Current workspace, found in request path
        self._current_workspace = None  # type: Workspace

        # Candidate workspace found in request body
        self._candidate_workspace = None  # type: Workspace

        # Authenticated user
        self._current_user = None  # type: User

        # User found from request headers, content, distinct from authenticated
        # user
        self._candidate_user = None  # type: User

        # TODO BS 2018-09-12: Must be deleted, see:
        # https://github.com/tracim/tracim/issues/903
        self.response_content_disposition = None
        # INFO - G.M - 18-05-2018 - Close db at the end of the request
        self.add_finished_callback(self._cleanup)
        self.add_response_callback(self._set_responses_headers)

    @property
    def current_workspace(self) -> Workspace:
        """
        Get current workspace of the request according to authentification and
        request headers (to retrieve workspace). Setted by default value the
        first time if not configured.
        :return: Workspace of the request
        """
        if self._current_workspace is None:
            self._current_workspace = self._get_current_workspace(self.current_user, self)   # nopep8
        return self._current_workspace

    @current_workspace.setter
    def current_workspace(self, workspace: Workspace) -> None:
        """
        Setting current_workspace
        :param workspace:
        :return:
        """
        if self._current_workspace is not None:
            raise ImmutableAttribute(
                "Can't modify already setted current_workspace"
            )
        self._current_workspace = workspace

    @property
    def current_user(self) -> User:
        """
        Get user from authentication mecanism.
        """
        if self._current_user is None:
            self.current_user = self._get_auth_safe_user(self)
        return self._current_user

    @current_user.setter
    def current_user(self, user: User) -> None:
        if self._current_user is not None:
            raise ImmutableAttribute(
                "Can't modify already setted current_user"
            )
        self._current_user = user

    @property
    def current_content(self) -> Content:
        """
        Get current  content from path
        """
        if self._current_content is None:
            self._current_content = self._get_current_content(
                self.current_user,
                self.current_workspace,
                self
                )
        return self._current_content

    @current_content.setter
    def current_content(self, content: Content) -> None:
        if self._current_content is not None:
            raise ImmutableAttribute(
                "Can't modify already setted current_content"
            )
        self._current_content = content

    @property
    def current_comment(self) -> Content:
        """
        Get current comment from path
        """
        if self._current_comment is None:
            self._current_comment = self._get_current_comment(
                self.current_user,
                self.current_workspace,
                self.current_content,
                self
                )
        return self._current_comment

    @current_comment.setter
    def current_comment(self, content: Content) -> None:
        if self._current_comment is not None:
            raise ImmutableAttribute(
                "Can't modify already setted current_content"
            )
        self._current_comment = content
    # TODO - G.M - 24-05-2018 - Find a better naming for this ?

    @property
    def candidate_user(self) -> User:
        """
        Get user from headers/body request. This user is not
        the one found by authentication mecanism. This user
        can help user to know about who one page is about in
        a similar way as current_workspace.
        """
        if self._candidate_user is None:
            self.candidate_user = self._get_candidate_user(self)
        return self._candidate_user

    @property
    def candidate_workspace(self) -> Workspace:
        """
        Get workspace from headers/body request. This workspace is not
        the one found from path. Its the one from json body.
        """
        if self._candidate_workspace is None:
            self._candidate_workspace = self._get_candidate_workspace(
                self.current_user,
                self
            )
        return self._candidate_workspace

    def response_download_mode(self, force_download: bool = False) -> None:
        if force_download:
            self.response_content_disposition = 'attachment'

    def _cleanup(self, request: 'TracimRequest') -> None:
        """
        Close dbsession at the end of the request in order to avoid exception
        about not properly closed session or "object created in another thread"
        issue
        see https://github.com/tracim/tracim_backend/issues/62
        :param request: same as self, request
        :return: nothing.
        """
        self._current_user = None
        self._current_workspace = None
        self.dbsession.close()

    def _set_responses_headers(self, request: 'TracimRequest', response: Response):  # nopep8
        if request.response_content_disposition:
            response.content_disposition = request.response_content_disposition

    @candidate_user.setter
    def candidate_user(self, user: User) -> None:
        if self._candidate_user is not None:
            raise ImmutableAttribute(
                "Can't modify already setted candidate_user"
            )
        self._candidate_user = user

    ###
    # Utils for TracimRequest
    ###
    def _get_current_comment(
            self,
            user: User,
            workspace: Workspace,
            content: Content,
            request: 'TracimRequest'
    ) -> Content:
        """
        Get current content from request
        :param user: User who want to check the workspace
        :param workspace: Workspace of the content
        :param content: comment is related to this content
        :param request: pyramid request
        :return: current content
        """
        comment_id = ''
        try:
            if 'comment_id' in request.matchdict:
                comment_id_str = request.matchdict['content_id']
                if not isinstance(comment_id_str, str) or not comment_id_str.isdecimal():  # nopep8
                    raise InvalidCommentId('comment_id is not a correct integer')  # nopep8
                comment_id = int(request.matchdict['comment_id'])
            if not comment_id:
                raise ContentNotFoundInTracimRequest('No comment_id property found in request')  # nopep8
            api = ContentApi(
                current_user=user,
                session=request.dbsession,
                show_deleted=True,
                show_archived=True,
                config=request.registry.settings['CFG']
            )
            comment = api.get_one(
                comment_id,
                content_type=CONTENT_TYPES.Comment.slug,
                workspace=workspace,
                parent=content,
            )
        except NoResultFound as exc:
            raise ContentNotFound(
                'Comment {} does not exist '
                'or is not visible for this user'.format(comment_id)
            ) from exc
        return comment

    def _get_current_content(
            self,
            user: User,
            workspace: Workspace,
            request: 'TracimRequest'
    ) -> Content:
        """
        Get current content from request
        :param user: User who want to check the workspace
        :param workspace: Workspace of the content
        :param request: pyramid request
        :return: current content
        """
        content_id = ''
        try:
            if 'content_id' in request.matchdict:
                content_id_str = request.matchdict['content_id']
                if not isinstance(content_id_str, str) or not content_id_str.isdecimal():  # nopep8
                    raise InvalidContentId('content_id is not a correct integer')  # nopep8
                content_id = int(request.matchdict['content_id'])
            if not content_id:
                raise ContentNotFoundInTracimRequest('No content_id property found in request')  # nopep8
            api = ContentApi(
                current_user=user,
                show_deleted=True,
                show_archived=True,
                session=request.dbsession,
                config=request.registry.settings['CFG']
            )
            content = api.get_one(content_id=content_id, workspace=workspace, content_type=CONTENT_TYPES.Any_SLUG)  # nopep8
        except NoResultFound as exc:
            raise ContentNotFound(
                'Content {} does not exist '
                'or is not visible for this user'.format(content_id)
            ) from exc
        return content

    def _get_candidate_user(
            self,
            request: 'TracimRequest',
    ) -> User:
        """
        Get candidate user
        :param request: pyramid request
        :return: user found from header/body
        """
        app_config = request.registry.settings['CFG']
        uapi = UserApi(None, show_deleted=True, session=request.dbsession, config=app_config)
        login = ''
        try:
            login = None
            if 'user_id' in request.matchdict:
                user_id_str = request.matchdict['user_id']
                if not isinstance(user_id_str, str) or not user_id_str.isdecimal():
                    raise InvalidUserId('user_id is not a correct integer')  # nopep8
                login = int(request.matchdict['user_id'])
            if not login:
                raise UserNotFoundInTracimRequest('You request a candidate user but the context not permit to found one')  # nopep8
            user = uapi.get_one(login)
        except UserNotFoundInTracimRequest as exc:
            raise UserDoesNotExist('User {} not found'.format(login)) from exc
        return user

    def _get_auth_safe_user(
            self,
            request: 'TracimRequest',
    ) -> User:
        """
        Get current pyramid authenticated user from request
        :param request: pyramid request
        :return: current authenticated user
        """
        app_config = request.registry.settings['CFG']
        uapi = UserApi(None, session=request.dbsession, config=app_config)
        login = ''
        try:
            login = request.authenticated_userid
            if not login:
                raise UserNotFoundInTracimRequest('You request a current user but the context not permit to found one')  # nopep8
            user = uapi.get_one_by_email(login)
            if not user.is_active:
                raise UserNotActive('User {} is not active'.format(login))
        except (UserDoesNotExist, UserNotFoundInTracimRequest) as exc:
            raise NotAuthenticated('User {} not found'.format(login)) from exc
        return user

    def _get_current_workspace(
            self,
            user: User,
            request: 'TracimRequest'
    ) -> Workspace:
        """
        Get current workspace from request
        :param user: User who want to check the workspace
        :param request: pyramid request
        :return: current workspace
        """
        workspace_id = ''
        try:
            if 'workspace_id' in request.matchdict:
                workspace_id_str = request.matchdict['workspace_id']
                if not isinstance(workspace_id_str, str) or not workspace_id_str.isdecimal():  # nopep8
                    raise InvalidWorkspaceId('workspace_id is not a correct integer')  # nopep8
                workspace_id = int(request.matchdict['workspace_id'])
            if not workspace_id:
                raise WorkspaceNotFoundInTracimRequest('No workspace_id property found in request')  # nopep8
            wapi = WorkspaceApi(
                current_user=user,
                session=request.dbsession,
                config=request.registry.settings['CFG'],
                show_deleted=True,
            )
            workspace = wapi.get_one(workspace_id)
        except NoResultFound as exc:
            raise WorkspaceNotFound(
                'Workspace {} does not exist '
                'or is not visible for this user'.format(workspace_id)
            ) from exc
        return workspace

    def _get_candidate_workspace(
            self,
            user: User,
            request: 'TracimRequest'
    ) -> Workspace:
        """
        Get current workspace from request
        :param user: User who want to check the workspace
        :param request: pyramid request
        :return: current workspace
        """
        workspace_id = ''
        try:
            if 'new_workspace_id' in request.json_body:
                workspace_id = request.json_body['new_workspace_id']
                if not isinstance(workspace_id, int):
                    if workspace_id.isdecimal():
                        workspace_id = int(workspace_id)
                    else:
                        raise InvalidWorkspaceId('workspace_id is not a correct integer')  # nopep8
            if not workspace_id:
                raise WorkspaceNotFoundInTracimRequest('No new_workspace_id property found in body')  # nopep8
            wapi = WorkspaceApi(
                current_user=user,
                session=request.dbsession,
                config=request.registry.settings['CFG'],
                show_deleted=True,
            )
            workspace = wapi.get_one(workspace_id)
        except JSONDecodeError as exc:
            raise WorkspaceNotFound('Invalid JSON content') from exc
        except NoResultFound as exc:
            raise WorkspaceNotFound(
                'Workspace {} does not exist '
                'or is not visible for this user'.format(workspace_id)
            ) from exc
        return workspace


1			# -- coding: utf-8 --
2			from pyramid.request import Request
3			from pyramid.response import Response
4			from sqlalchemy.orm.exc import NoResultFound
5
6			from tracim_backend.exceptions import NotAuthenticated
7			from tracim_backend.exceptions import UserNotActive
8			from tracim_backend.exceptions import ContentNotFound
9			from tracim_backend.exceptions import InvalidUserId
10			from tracim_backend.exceptions import InvalidWorkspaceId
11			from tracim_backend.exceptions import InvalidContentId
12			from tracim_backend.exceptions import InvalidCommentId
13			from tracim_backend.exceptions import ContentNotFoundInTracimRequest
14			from tracim_backend.exceptions import WorkspaceNotFoundInTracimRequest
15			from tracim_backend.exceptions import UserNotFoundInTracimRequest
16			from tracim_backend.exceptions import UserDoesNotExist
17			from tracim_backend.exceptions import WorkspaceNotFound
18			from tracim_backend.exceptions import ImmutableAttribute
19			from tracim_backend.app_models.contents import CONTENT_TYPES
20			from tracim_backend.lib.core.content import ContentApi
21			from tracim_backend.lib.core.user import UserApi
22			from tracim_backend.lib.core.workspace import WorkspaceApi
23			from tracim_backend.lib.utils.authorization import JSONDecodeError
24
25			from tracim_backend.models import User
26			from tracim_backend.models.data import Workspace
27			from tracim_backend.models.data import Content
28
29
30			class TracimRequest(Request):
31			"""
32			Request with tracim specific params/methods
33			"""
34			def __init__(
35			self,
36			environ,
37			charset=None,
38			unicode_errors=None,
39			decode_param_names=None,
40			**kw
41			):
42			super().__init__(
43			environ,
44			charset,
45			unicode_errors,
46			decode_param_names,
47			**kw
48			)
49			# Current comment, found in request path
50			self._current_comment = None # type: Content
51
52			# Current content, found in request path
53			self._current_content = None # type: Content
54
55			# Current workspace, found in request path
56			self._current_workspace = None # type: Workspace
57
58			# Candidate workspace found in request body
59			self._candidate_workspace = None # type: Workspace
60
61			# Authenticated user
62			self._current_user = None # type: User
63
64			# User found from request headers, content, distinct from authenticated
65			# user
66			self._candidate_user = None # type: User
67
68			# TODO BS 2018-09-12: Must be deleted, see:
69			# https://github.com/tracim/tracim/issues/903
70			self.response_content_disposition = None
71			# INFO - G.M - 18-05-2018 - Close db at the end of the request
72			self.add_finished_callback(self._cleanup)
73			self.add_response_callback(self._set_responses_headers)
74
75			@property
76			def current_workspace(self) -> Workspace:
77			"""
78			Get current workspace of the request according to authentification and
79			request headers (to retrieve workspace). Setted by default value the
80			first time if not configured.
81			:return: Workspace of the request
82			"""
83			if self._current_workspace is None:
84			self._current_workspace = self._get_current_workspace(self.current_user, self) # nopep8
85			return self._current_workspace
86
87			@current_workspace.setter
88			def current_workspace(self, workspace: Workspace) -> None:
89			"""
90			Setting current_workspace
91			:param workspace:
92			:return:
93			"""
94			if self._current_workspace is not None:
95			raise ImmutableAttribute(
96			"Can't modify already setted current_workspace"
97			)
98			self._current_workspace = workspace
99
100			@property
101			def current_user(self) -> User:
102			"""
103			Get user from authentication mecanism.
104			"""
105			if self._current_user is None:
106			self.current_user = self._get_auth_safe_user(self)
107			return self._current_user
108
109			@current_user.setter
110			def current_user(self, user: User) -> None:
111			if self._current_user is not None:
112			raise ImmutableAttribute(
113			"Can't modify already setted current_user"
114			)
115			self._current_user = user
116
117			@property
118			def current_content(self) -> Content:
119			"""
120			Get current content from path
121			"""
122			if self._current_content is None:
123			self._current_content = self._get_current_content(
124			self.current_user,
125			self.current_workspace,
126			self
127			)
128			return self._current_content
129
130			@current_content.setter
131			def current_content(self, content: Content) -> None:
132			if self._current_content is not None:
133			raise ImmutableAttribute(
134			"Can't modify already setted current_content"
135			)
136			self._current_content = content
137
138			@property
139			def current_comment(self) -> Content:
140			"""
141			Get current comment from path
142			"""
143			if self._current_comment is None:
144			self._current_comment = self._get_current_comment(
145			self.current_user,
146			self.current_workspace,
147			self.current_content,
148			self
149			)
150			return self._current_comment
151
152			@current_comment.setter
153			def current_comment(self, content: Content) -> None:
154			if self._current_comment is not None:
155			raise ImmutableAttribute(
156			"Can't modify already setted current_content"
157			)
158			self._current_comment = content
159			# TODO - G.M - 24-05-2018 - Find a better naming for this ?
160
161			@property
162			def candidate_user(self) -> User:
163			"""
164			Get user from headers/body request. This user is not
165			the one found by authentication mecanism. This user
166			can help user to know about who one page is about in
167			a similar way as current_workspace.
168			"""
169			if self._candidate_user is None:
170			self.candidate_user = self._get_candidate_user(self)
171			return self._candidate_user
172
173			@property
174			def candidate_workspace(self) -> Workspace:
175			"""
176			Get workspace from headers/body request. This workspace is not
177			the one found from path. Its the one from json body.
178			"""
179			if self._candidate_workspace is None:
180			self._candidate_workspace = self._get_candidate_workspace(
181			self.current_user,
182			self
183			)
184			return self._candidate_workspace
185
186			def response_download_mode(self, force_download: bool = False) -> None:
187			if force_download:
188			self.response_content_disposition = 'attachment'
189
190			def _cleanup(self, request: 'TracimRequest') -> None:
191			"""
192			Close dbsession at the end of the request in order to avoid exception
193			about not properly closed session or "object created in another thread"
194			issue
195			see https://github.com/tracim/tracim_backend/issues/62
196			:param request: same as self, request
197			:return: nothing.
198			"""
199			self._current_user = None
200			self._current_workspace = None
201			self.dbsession.close()
202
203			def _set_responses_headers(self, request: 'TracimRequest', response: Response): # nopep8
204			if request.response_content_disposition:
205			response.content_disposition = request.response_content_disposition
206
207			@candidate_user.setter
208			def candidate_user(self, user: User) -> None:
209			if self._candidate_user is not None:
210			raise ImmutableAttribute(
211			"Can't modify already setted candidate_user"
212			)
213			self._candidate_user = user
214
215			###
216			# Utils for TracimRequest
217			###
218			def _get_current_comment(
219			self,
220			user: User,
221			workspace: Workspace,
222			content: Content,
223			request: 'TracimRequest'
224			) -> Content:
225			"""
226			Get current content from request
227			:param user: User who want to check the workspace
228			:param workspace: Workspace of the content
229			:param content: comment is related to this content
230			:param request: pyramid request
231			:return: current content
232			"""
233			comment_id = ''
234			try:
235			if 'comment_id' in request.matchdict:
236			comment_id_str = request.matchdict['content_id']
237			if not isinstance(comment_id_str, str) or not comment_id_str.isdecimal(): # nopep8
238			raise InvalidCommentId('comment_id is not a correct integer') # nopep8
239			comment_id = int(request.matchdict['comment_id'])
240			if not comment_id:
241			raise ContentNotFoundInTracimRequest('No comment_id property found in request') # nopep8
242			api = ContentApi(
243			current_user=user,
244			session=request.dbsession,
245			show_deleted=True,
246			show_archived=True,
247			config=request.registry.settings['CFG']
248			)
249			comment = api.get_one(
250			comment_id,
251			content_type=CONTENT_TYPES.Comment.slug,
252			workspace=workspace,
253			parent=content,
254			)
255			except NoResultFound as exc:
256			raise ContentNotFound(
257			'Comment {} does not exist '
258			'or is not visible for this user'.format(comment_id)
259			) from exc
260			return comment
261
262			def _get_current_content(
263			self,
264			user: User,
265			workspace: Workspace,
266			request: 'TracimRequest'
267			) -> Content:
268			"""
269			Get current content from request
270			:param user: User who want to check the workspace
271			:param workspace: Workspace of the content
272			:param request: pyramid request
273			:return: current content
274			"""
275			content_id = ''
276			try:
277			if 'content_id' in request.matchdict:
278			content_id_str = request.matchdict['content_id']
279			if not isinstance(content_id_str, str) or not content_id_str.isdecimal(): # nopep8
280			raise InvalidContentId('content_id is not a correct integer') # nopep8
281			content_id = int(request.matchdict['content_id'])
282			if not content_id:
283			raise ContentNotFoundInTracimRequest('No content_id property found in request') # nopep8
284			api = ContentApi(
285			current_user=user,
286			show_deleted=True,
287			show_archived=True,
288			session=request.dbsession,
289			config=request.registry.settings['CFG']
290			)
291			content = api.get_one(content_id=content_id, workspace=workspace, content_type=CONTENT_TYPES.Any_SLUG) # nopep8
292			except NoResultFound as exc:
293			raise ContentNotFound(
294			'Content {} does not exist '
295			'or is not visible for this user'.format(content_id)
296			) from exc
297			return content
298
299			def _get_candidate_user(
300			self,
301			request: 'TracimRequest',
302			) -> User:
303			"""
304			Get candidate user
305			:param request: pyramid request
306			:return: user found from header/body
307			"""
308			app_config = request.registry.settings['CFG']
309			uapi = UserApi(None, show_deleted=True, session=request.dbsession, config=app_config)
310			login = ''
311			try:
312			login = None
313			if 'user_id' in request.matchdict:
314			user_id_str = request.matchdict['user_id']
315			if not isinstance(user_id_str, str) or not user_id_str.isdecimal():
316			raise InvalidUserId('user_id is not a correct integer') # nopep8
317			login = int(request.matchdict['user_id'])
318			if not login:
319			raise UserNotFoundInTracimRequest('You request a candidate user but the context not permit to found one') # nopep8
320			user = uapi.get_one(login)
321			except UserNotFoundInTracimRequest as exc:
322			raise UserDoesNotExist('User {} not found'.format(login)) from exc
323			return user
324
325			def _get_auth_safe_user(
326			self,
327			request: 'TracimRequest',
328			) -> User:
329			"""
330			Get current pyramid authenticated user from request
331			:param request: pyramid request
332			:return: current authenticated user
333			"""
334			app_config = request.registry.settings['CFG']
335			uapi = UserApi(None, session=request.dbsession, config=app_config)
336			login = ''
337			try:
338			login = request.authenticated_userid
339			if not login:
340			raise UserNotFoundInTracimRequest('You request a current user but the context not permit to found one') # nopep8
341			user = uapi.get_one_by_email(login)
342			if not user.is_active:
343			raise UserNotActive('User {} is not active'.format(login))
344			except (UserDoesNotExist, UserNotFoundInTracimRequest) as exc:
345			raise NotAuthenticated('User {} not found'.format(login)) from exc
346			return user
347
348			def _get_current_workspace(
349			self,
350			user: User,
351			request: 'TracimRequest'
352			) -> Workspace:
353			"""
354			Get current workspace from request
355			:param user: User who want to check the workspace
356			:param request: pyramid request
357			:return: current workspace
358			"""
359			workspace_id = ''
360			try:
361			if 'workspace_id' in request.matchdict:
362			workspace_id_str = request.matchdict['workspace_id']
363			if not isinstance(workspace_id_str, str) or not workspace_id_str.isdecimal(): # nopep8
364			raise InvalidWorkspaceId('workspace_id is not a correct integer') # nopep8
365			workspace_id = int(request.matchdict['workspace_id'])
366			if not workspace_id:
367			raise WorkspaceNotFoundInTracimRequest('No workspace_id property found in request') # nopep8
368			wapi = WorkspaceApi(
369			current_user=user,
370			session=request.dbsession,
371			config=request.registry.settings['CFG'],
372			show_deleted=True,
373			)
374			workspace = wapi.get_one(workspace_id)
375			except NoResultFound as exc:
376			raise WorkspaceNotFound(
377			'Workspace {} does not exist '
378			'or is not visible for this user'.format(workspace_id)
379			) from exc
380			return workspace
381
382			def _get_candidate_workspace(
383			self,
384			user: User,
385			request: 'TracimRequest'
386			) -> Workspace:
387			"""
388			Get current workspace from request
389			:param user: User who want to check the workspace
390			:param request: pyramid request
391			:return: current workspace
392			"""
393			workspace_id = ''
394			try:
395			if 'new_workspace_id' in request.json_body:
396			workspace_id = request.json_body['new_workspace_id']
397			if not isinstance(workspace_id, int):
398			if workspace_id.isdecimal():
399			workspace_id = int(workspace_id)
400			else:
401			raise InvalidWorkspaceId('workspace_id is not a correct integer') # nopep8
402			if not workspace_id:
403			raise WorkspaceNotFoundInTracimRequest('No new_workspace_id property found in body') # nopep8
404			wapi = WorkspaceApi(
405			current_user=user,
406			session=request.dbsession,
407			config=request.registry.settings['CFG'],
408			show_deleted=True,
409			)
410			workspace = wapi.get_one(workspace_id)
411			except JSONDecodeError as exc:
412			raise WorkspaceNotFound('Invalid JSON content') from exc
413			except NoResultFound as exc:
414			raise WorkspaceNotFound(
415			'Workspace {} does not exist '
416			'or is not visible for this user'.format(workspace_id)
417			) from exc
418			return workspace
419

tracim / tracim_v2

Push — develop ( c8e157...770c39 )

TracimRequest._cleanup() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like