user.py 3.6KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126
  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. class UserApi(object):
  10. def __init__(self, current_user: typing.Optional[User], session, config):
  11. self._session = session
  12. self._user = current_user
  13. self._config = config
  14. def _base_query(self):
  15. return self._session.query(User)
  16. def get_one(self, user_id: int) -> User:
  17. return self._base_query().filter(User.user_id==user_id).one()
  18. def get_one_by_email(self, email: str) -> User:
  19. return self._base_query().filter(User.email==email).one()
  20. def get_one_by_id(self, id: int) -> User:
  21. return self._base_query().filter(User.user_id==id).one()
  22. def get_all(self) -> typing.Iterable[User]:
  23. return self._session.query(User).order_by(User.display_name).all()
  24. def user_with_email_exists(self, email: str):
  25. try:
  26. self.get_one_by_email(email)
  27. return True
  28. # TODO - G.M - 09-04-2018 - Better exception
  29. except:
  30. return False
  31. def authenticate_user(self, email, password) -> User:
  32. """
  33. Authenticate user with email and password, raise AuthenticationFailed
  34. if uncorrect.
  35. :param email: email of the user
  36. :param password: cleartext password of the user
  37. :return: User who was authenticated.
  38. """
  39. try:
  40. user = self.get_one_by_email(email)
  41. if user.validate_password(password):
  42. return user
  43. else:
  44. raise BadUserPassword()
  45. except (BadUserPassword, NoResultFound):
  46. raise AuthenticationFailed
  47. def update(
  48. self,
  49. user: User,
  50. name: str=None,
  51. email: str=None,
  52. do_save=True,
  53. timezone: str='',
  54. ):
  55. if name is not None:
  56. user.display_name = name
  57. if email is not None:
  58. user.email = email
  59. user.timezone = timezone
  60. if do_save:
  61. self.save(user)
  62. def create_user(self, email=None, groups=[], save_now=False) -> User:
  63. user = User()
  64. if email:
  65. user.email = email
  66. for group in groups:
  67. user.groups.append(group)
  68. self._session.add(user)
  69. if save_now:
  70. self._session.flush()
  71. return user
  72. def save(self, user: User):
  73. self._session.flush()
  74. def execute_created_user_actions(self, created_user: User) -> None:
  75. """
  76. Execute actions when user just been created
  77. :return:
  78. """
  79. # NOTE: Cyclic import
  80. # TODO - G.M - 28-03-2018 - [Calendar] Reenable Calendar stuff
  81. #from tracim.lib.calendar import CalendarManager
  82. #from tracim.model.organisational import UserCalendar
  83. # TODO - G.M - 04-04-2018 - [auth]
  84. # Check if this is already needed with
  85. # new auth system
  86. created_user.ensure_auth_token(
  87. session=self._session,
  88. validity_seconds=self._config.USER_AUTH_TOKEN_VALIDITY
  89. )
  90. # Ensure database is up-to-date
  91. self._session.flush()
  92. transaction.commit()
  93. # TODO - G.M - 28-03-2018 - [Calendar] Reenable Calendar stuff
  94. # calendar_manager = CalendarManager(created_user)
  95. # calendar_manager.create_then_remove_fake_event(
  96. # calendar_class=UserCalendar,
  97. # related_object_id=created_user.user_id,
  98. # )