user.py 4.9KB

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