TracimRequest._get_candidate_workspace()   B
last analyzed

Complexity

Conditions 7

Size

Total Lines 37
Code Lines 28

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 28
dl 0
loc 37
rs 7.808
c 0
b 0
f 0
cc 7
nop 3
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 UserAuthenticatedIsNotActive
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_type_list
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
        # INFO - G.M - 18-05-2018 - Close db at the end of the request
69
        self.add_finished_callback(self._cleanup)
70
71
    @property
72
    def current_workspace(self) -> Workspace:
73
        """
74
        Get current workspace of the request according to authentification and
75
        request headers (to retrieve workspace). Setted by default value the
76
        first time if not configured.
77
        :return: Workspace of the request
78
        """
79
        if self._current_workspace is None:
80
            self._current_workspace = self._get_current_workspace(self.current_user, self)   # nopep8
81
        return self._current_workspace
82
83
    @current_workspace.setter
84
    def current_workspace(self, workspace: Workspace) -> None:
85
        """
86
        Setting current_workspace
87
        :param workspace:
88
        :return:
89
        """
90
        if self._current_workspace is not None:
91
            raise ImmutableAttribute(
92
                "Can't modify already setted current_workspace"
93
            )
94
        self._current_workspace = workspace
95
96
    @property
97
    def current_user(self) -> User:
98
        """
99
        Get user from authentication mecanism.
100
        """
101
        if self._current_user is None:
102
            self.current_user = self._get_auth_safe_user(self)
103
        return self._current_user
104
105
    @current_user.setter
106
    def current_user(self, user: User) -> None:
107
        if self._current_user is not None:
108
            raise ImmutableAttribute(
109
                "Can't modify already setted current_user"
110
            )
111
        self._current_user = user
112
113
    @property
114
    def current_content(self) -> Content:
115
        """
116
        Get current  content from path
117
        """
118
        if self._current_content is None:
119
            self._current_content = self._get_current_content(
120
                self.current_user,
121
                self.current_workspace,
122
                self
123
                )
124
        return self._current_content
125
126
    @current_content.setter
127
    def current_content(self, content: Content) -> None:
128
        if self._current_content is not None:
129
            raise ImmutableAttribute(
130
                "Can't modify already setted current_content"
131
            )
132
        self._current_content = content
133
134
    @property
135
    def current_comment(self) -> Content:
136
        """
137
        Get current comment from path
138
        """
139
        if self._current_comment is None:
140
            self._current_comment = self._get_current_comment(
141
                self.current_user,
142
                self.current_workspace,
143
                self.current_content,
144
                self
145
                )
146
        return self._current_comment
147
148
    @current_comment.setter
149
    def current_comment(self, content: Content) -> None:
150
        if self._current_comment is not None:
151
            raise ImmutableAttribute(
152
                "Can't modify already setted current_content"
153
            )
154
        self._current_comment = content
155
    # TODO - G.M - 24-05-2018 - Find a better naming for this ?
156
157
    @property
158
    def candidate_user(self) -> User:
159
        """
160
        Get user from headers/body request. This user is not
161
        the one found by authentication mecanism. This user
162
        can help user to know about who one page is about in
163
        a similar way as current_workspace.
164
        """
165
        if self._candidate_user is None:
166
            self.candidate_user = self._get_candidate_user(self)
167
        return self._candidate_user
168
169
    @property
170
    def candidate_workspace(self) -> Workspace:
171
        """
172
        Get workspace from headers/body request. This workspace is not
173
        the one found from path. Its the one from json body.
174
        """
175
        if self._candidate_workspace is None:
176
            self._candidate_workspace = self._get_candidate_workspace(
177
                self.current_user,
178
                self
179
            )
180
        return self._candidate_workspace
181
182
    def _cleanup(self, request: 'TracimRequest') -> None:
183
        """
184
        Close dbsession at the end of the request in order to avoid exception
185
        about not properly closed session or "object created in another thread"
186
        issue
187
        see https://github.com/tracim/tracim_backend/issues/62
188
        :param request: same as self, request
189
        :return: nothing.
190
        """
191
        self._current_user = None
192
        self._current_workspace = None
193
        self.dbsession.close()
194
195
    @candidate_user.setter
196
    def candidate_user(self, user: User) -> None:
197
        if self._candidate_user is not None:
198
            raise ImmutableAttribute(
199
                "Can't modify already setted candidate_user"
200
            )
201
        self._candidate_user = user
202
203
    ###
204
    # Utils for TracimRequest
205
    ###
206
    def _get_current_comment(
207
            self,
208
            user: User,
209
            workspace: Workspace,
210
            content: Content,
211
            request: 'TracimRequest'
212
    ) -> Content:
213
        """
214
        Get current content from request
215
        :param user: User who want to check the workspace
216
        :param workspace: Workspace of the content
217
        :param content: comment is related to this content
218
        :param request: pyramid request
219
        :return: current content
220
        """
221
        comment_id = ''
222
        try:
223
            if 'comment_id' in request.matchdict:
224
                comment_id_str = request.matchdict['content_id']
225
                if not isinstance(comment_id_str, str) or not comment_id_str.isdecimal():  # nopep8
226
                    raise InvalidCommentId('comment_id is not a correct integer')  # nopep8
227
                comment_id = int(request.matchdict['comment_id'])
228
            if not comment_id:
229
                raise ContentNotFoundInTracimRequest('No comment_id property found in request')  # nopep8
230
            api = ContentApi(
231
                current_user=user,
232
                session=request.dbsession,
233
                show_deleted=True,
234
                show_archived=True,
235
                config=request.registry.settings['CFG']
236
            )
237
            comment = api.get_one(
238
                comment_id,
239
                content_type=content_type_list.Comment.slug,
240
                workspace=workspace,
241
                parent=content,
242
            )
243
        except NoResultFound as exc:
244
            raise ContentNotFound(
245
                'Comment {} does not exist '
246
                'or is not visible for this user'.format(comment_id)
247
            ) from exc
248
        return comment
249
250
    def _get_current_content(
251
            self,
252
            user: User,
253
            workspace: Workspace,
254
            request: 'TracimRequest'
255
    ) -> Content:
256
        """
257
        Get current content from request
258
        :param user: User who want to check the workspace
259
        :param workspace: Workspace of the content
260
        :param request: pyramid request
261
        :return: current content
262
        """
263
        content_id = ''
264
        try:
265
            if 'content_id' in request.matchdict:
266
                content_id_str = request.matchdict['content_id']
267
                if not isinstance(content_id_str, str) or not content_id_str.isdecimal():  # nopep8
268
                    raise InvalidContentId('content_id is not a correct integer')  # nopep8
269
                content_id = int(request.matchdict['content_id'])
270
            if not content_id:
271
                raise ContentNotFoundInTracimRequest('No content_id property found in request')  # nopep8
272
            api = ContentApi(
273
                current_user=user,
274
                show_deleted=True,
275
                show_archived=True,
276
                session=request.dbsession,
277
                config=request.registry.settings['CFG']
278
            )
279
            content = api.get_one(content_id=content_id, workspace=workspace, content_type=content_type_list.Any_SLUG)  # nopep8
280
        except NoResultFound as exc:
281
            raise ContentNotFound(
282
                'Content {} does not exist '
283
                'or is not visible for this user'.format(content_id)
284
            ) from exc
285
        return content
286
287
    def _get_candidate_user(
288
            self,
289
            request: 'TracimRequest',
290
    ) -> User:
291
        """
292
        Get candidate user
293
        :param request: pyramid request
294
        :return: user found from header/body
295
        """
296
        app_config = request.registry.settings['CFG']
297
        uapi = UserApi(None, show_deleted=True, session=request.dbsession, config=app_config)
298
        login = ''
299
        try:
300
            login = None
301
            if 'user_id' in request.matchdict:
302
                user_id_str = request.matchdict['user_id']
303
                if not isinstance(user_id_str, str) or not user_id_str.isdecimal():
304
                    raise InvalidUserId('user_id is not a correct integer')  # nopep8
305
                login = int(request.matchdict['user_id'])
306
            if not login:
307
                raise UserNotFoundInTracimRequest('You request a candidate user but the context not permit to found one')  # nopep8
308
            user = uapi.get_one(login)
309
        except UserNotFoundInTracimRequest as exc:
310
            raise UserDoesNotExist('User {} not found'.format(login)) from exc
311
        return user
312
313
    def _get_auth_safe_user(
314
            self,
315
            request: 'TracimRequest',
316
    ) -> User:
317
        """
318
        Get current pyramid authenticated user from request
319
        :param request: pyramid request
320
        :return: current authenticated user
321
        """
322
        app_config = request.registry.settings['CFG']
323
        uapi = UserApi(None, session=request.dbsession, config=app_config)
324
        login = ''
325
        try:
326
            login = request.authenticated_userid
327
            if not login:
328
                raise UserNotFoundInTracimRequest('You request a current user but the context not permit to found one')  # nopep8
329
            user = uapi.get_one(login)
330
            if not user.is_active:
331
                raise UserAuthenticatedIsNotActive('User {} is not active'.format(login))
332
        except (UserDoesNotExist, UserNotFoundInTracimRequest) as exc:
333
            raise NotAuthenticated('User {} not found'.format(login)) from exc
334
        return user
335
336
    def _get_current_workspace(
337
            self,
338
            user: User,
339
            request: 'TracimRequest'
340
    ) -> Workspace:
341
        """
342
        Get current workspace from request
343
        :param user: User who want to check the workspace
344
        :param request: pyramid request
345
        :return: current workspace
346
        """
347
        workspace_id = ''
348
        try:
349
            if 'workspace_id' in request.matchdict:
350
                workspace_id_str = request.matchdict['workspace_id']
351
                if not isinstance(workspace_id_str, str) or not workspace_id_str.isdecimal():  # nopep8
352
                    raise InvalidWorkspaceId('workspace_id is not a correct integer')  # nopep8
353
                workspace_id = int(request.matchdict['workspace_id'])
354
            if not workspace_id:
355
                raise WorkspaceNotFoundInTracimRequest('No workspace_id property found in request')  # nopep8
356
            wapi = WorkspaceApi(
357
                current_user=user,
358
                session=request.dbsession,
359
                config=request.registry.settings['CFG'],
360
                show_deleted=True,
361
            )
362
            workspace = wapi.get_one(workspace_id)
363
        except NoResultFound as exc:
364
            raise WorkspaceNotFound(
365
                'Workspace {} does not exist '
366
                'or is not visible for this user'.format(workspace_id)
367
            ) from exc
368
        return workspace
369
370
    def _get_candidate_workspace(
371
            self,
372
            user: User,
373
            request: 'TracimRequest'
374
    ) -> Workspace:
375
        """
376
        Get current workspace from request
377
        :param user: User who want to check the workspace
378
        :param request: pyramid request
379
        :return: current workspace
380
        """
381
        workspace_id = ''
382
        try:
383
            if 'new_workspace_id' in request.json_body:
384
                workspace_id = request.json_body['new_workspace_id']
385
                if not isinstance(workspace_id, int):
386
                    if workspace_id.isdecimal():
387
                        workspace_id = int(workspace_id)
388
                    else:
389
                        raise InvalidWorkspaceId('workspace_id is not a correct integer')  # nopep8
390
            if not workspace_id:
391
                raise WorkspaceNotFoundInTracimRequest('No new_workspace_id property found in body')  # nopep8
392
            wapi = WorkspaceApi(
393
                current_user=user,
394
                session=request.dbsession,
395
                config=request.registry.settings['CFG'],
396
                show_deleted=True,
397
            )
398
            workspace = wapi.get_one(workspace_id)
399
        except JSONDecodeError as exc:
400
            raise WorkspaceNotFound('Invalid JSON content') from exc
401
        except NoResultFound as exc:
402
            raise WorkspaceNotFound(
403
                'Workspace {} does not exist '
404
                'or is not visible for this user'.format(workspace_id)
405
            ) from exc
406
        return workspace
407