user.py 5.1KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183
  1. # -*- coding: utf-8 -*-
  2. import threading
  3. import transaction
  4. import typing as typing
  5. from tracim.models.auth import User
  6. from sqlalchemy.orm.exc import NoResultFound
  7. from tracim.exceptions import BadUserPassword
  8. from tracim.exceptions import AuthenticationFailed
  9. from tracim.models.context_models import UserInContext
  10. class UserApi(object):
  11. def __init__(self, current_user: typing.Optional[User], session, config):
  12. self._session = session
  13. self._user = current_user
  14. self._config = config
  15. def _base_query(self):
  16. return self._session.query(User)
  17. def _get_correct_user_type(
  18. self,
  19. user: User,
  20. in_context: bool,
  21. ) -> typing.Union[User, UserInContext]:
  22. """
  23. Choose user type object depending on in_context bool.
  24. :param user:
  25. :param in_context:
  26. :return: user as User or UserInContext if in_context is True
  27. """
  28. if in_context:
  29. user = UserInContext(
  30. user=user,
  31. dbsession=self._session,
  32. config=self._config,
  33. )
  34. return user
  35. # Getters
  36. def get_one(
  37. self,
  38. user_id: int,
  39. in_context: bool=False,
  40. ) -> typing.Union[UserInContext, User]:
  41. """
  42. Get one user by user id
  43. :param user_id:
  44. :param in_context: Return User or UserInContext Object
  45. :return: one user
  46. """
  47. user = self._base_query().filter(User.user_id == user_id).one()
  48. return self._get_correct_user_type(user, in_context)
  49. def get_one_by_email(
  50. self,
  51. email: str,
  52. in_context: bool=False,
  53. ) -> User:
  54. """
  55. Get one user by email
  56. :param email: Email of the user
  57. :param in_context: Return User or UserInContext Object
  58. :return: one user
  59. """
  60. user = self._base_query().filter(User.email == email).one()
  61. return self._get_correct_user_type(user, in_context)
  62. # FIXME - G.M - 24-04-2018 - Duplicate method with get_one.
  63. def get_one_by_id(self, id: int, in_context=False) -> User:
  64. return self.get_one(user_id=id, in_context=in_context)
  65. def get_current(self, in_context: bool=False):
  66. """
  67. Get current_user
  68. :param in_context:
  69. :return:
  70. """
  71. return self._get_correct_user_type(self._user, in_context)
  72. def get_all(self) -> typing.Iterable[User]:
  73. return self._session.query(User).order_by(User.display_name).all()
  74. # Check methods
  75. def user_with_email_exists(self, email: str):
  76. try:
  77. self.get_one_by_email(email)
  78. return True
  79. # TODO - G.M - 09-04-2018 - Better exception
  80. except:
  81. return False
  82. def authenticate_user(self, email, password) -> User:
  83. """
  84. Authenticate user with email and password, raise AuthenticationFailed
  85. if uncorrect.
  86. :param email: email of the user
  87. :param password: cleartext password of the user
  88. :return: User who was authenticated.
  89. """
  90. try:
  91. user = self.get_one_by_email(email)
  92. if user.validate_password(password):
  93. return user
  94. else:
  95. raise BadUserPassword()
  96. except (BadUserPassword, NoResultFound):
  97. raise AuthenticationFailed
  98. # Actions
  99. def update(
  100. self,
  101. user: User,
  102. name: str=None,
  103. email: str=None,
  104. do_save=True,
  105. timezone: str='',
  106. ):
  107. if name is not None:
  108. user.display_name = name
  109. if email is not None:
  110. user.email = email
  111. user.timezone = timezone
  112. if do_save:
  113. self.save(user)
  114. def create_user(self, email=None, groups=[], save_now=False) -> User:
  115. user = User()
  116. if email:
  117. user.email = email
  118. for group in groups:
  119. user.groups.append(group)
  120. self._session.add(user)
  121. if save_now:
  122. self._session.flush()
  123. return user
  124. def save(self, user: User):
  125. self._session.flush()
  126. def execute_created_user_actions(self, created_user: User) -> None:
  127. """
  128. Execute actions when user just been created
  129. :return:
  130. """
  131. # NOTE: Cyclic import
  132. # TODO - G.M - 28-03-2018 - [Calendar] Reenable Calendar stuff
  133. #from tracim.lib.calendar import CalendarManager
  134. #from tracim.model.organisational import UserCalendar
  135. # TODO - G.M - 04-04-2018 - [auth]
  136. # Check if this is already needed with
  137. # new auth system
  138. created_user.ensure_auth_token(
  139. session=self._session,
  140. validity_seconds=self._config.USER_AUTH_TOKEN_VALIDITY
  141. )
  142. # Ensure database is up-to-date
  143. self._session.flush()
  144. transaction.commit()
  145. # TODO - G.M - 28-03-2018 - [Calendar] Reenable Calendar stuff
  146. # calendar_manager = CalendarManager(created_user)
  147. # calendar_manager.create_then_remove_fake_event(
  148. # calendar_class=UserCalendar,
  149. # related_object_id=created_user.user_id,
  150. # )