tracim_v2/tracim/lib/utils/request.py

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

from tracim.exceptions import NotAuthenticated
from tracim.exceptions import UserNotFoundInTracimRequest
from tracim.exceptions import UserDoesNotExist
from tracim.exceptions import WorkspaceNotFound
from tracim.exceptions import ImmutableAttribute
from tracim.lib.core.user import UserApi
from tracim.lib.core.workspace import WorkspaceApi
from tracim.lib.utils.authorization import JSONDecodeError

from tracim.models import User
from tracim.models.data import Workspace


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 workspace, found by request headers or content
        self._current_workspace = None  # type: Workspace

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

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

        # INFO - G.M - 18-05-2018 - Close db at the end of the request
        self.add_finished_callback(self._cleanup)

    @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 = get_workspace(self.current_user, self)
        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 = 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

    # 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._user_candidate is None:
            self.candidate_user = get_candidate_user(self)
        return self._user_candidate

    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()


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


def get_candidate_user(
        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, session=request.dbsession, config=app_config)

    try:
        login = None
        if 'user_id' in request.matchdict:
            login = 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(
        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)
    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)
    except (UserDoesNotExist, UserNotFoundInTracimRequest) as exc:
        raise NotAuthenticated('User {} not found'.format(login)) from exc
    return user


def get_workspace(
        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 = request.matchdict['workspace_id']
        if not workspace_id:
            raise WorkspaceNotFound('No workspace_id property found in request')
        wapi = WorkspaceApi(
            current_user=user,
            session=request.dbsession,
            config=request.registry.settings['CFG']
        )
        workspace = wapi.get_one(workspace_id)
    except JSONDecodeError:
        raise WorkspaceNotFound('Bad json body')
    except NoResultFound:
        raise WorkspaceNotFound(
            'Workspace {} does not exist '
            'or is not visible for this user'.format(workspace_id)
        )
    return workspace