user.py 3.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889
  1. import logging
  2. import os
  3. from typing import Any, Callable, Dict
  4. from descope import AuthException, DescopeClient
  5. from nicegui import Client, app, ui
  6. _descope_id = os.environ.get('DESCOPE_PROJECT_ID', '')
  7. try:
  8. descope_client = DescopeClient(project_id=_descope_id)
  9. except Exception as error:
  10. logging.exception("failed to initialize.")
  11. def login_form() -> ui.element:
  12. with ui.card().classes('w-96 mx-auto'):
  13. return ui.element('descope-wc').props(f'project-id="{_descope_id}" flow-id="sign-up-or-in"') \
  14. .on('success', lambda e: app.storage.user.update({'descope': e.args['detail']['user']}))
  15. def about() -> Dict[str, Any]:
  16. try:
  17. return app.storage.user['descope']
  18. except AuthException:
  19. logging.exception("Unable to load user.")
  20. return {}
  21. async def logout() -> None:
  22. result = await ui.run_javascript('return await sdk.logout()', respond=True)
  23. if result['code'] != 200:
  24. logging.error(f'Logout failed: {result}')
  25. ui.notify('Logout failed', type='negative')
  26. else:
  27. app.storage.user['descope'] = None
  28. ui.open('/login')
  29. class page(ui.page):
  30. def __init__(self, path):
  31. super().__init__(path)
  32. def __call__(self, func: Callable[..., Any]) -> Callable[..., Any]:
  33. async def content(client: Client):
  34. ui.add_head_html('<script src="https://unpkg.com/@descope/web-component@latest/dist/index.js"></script>')
  35. ui.add_head_html('<script src="https://unpkg.com/@descope/web-js-sdk@latest/dist/index.umd.js"></script>')
  36. ui.add_body_html('''
  37. <script>
  38. const sdk = Descope({ projectId: \'''' + _descope_id + '''\', persistTokens: true, autoRefresh: true });
  39. const sessionToken = sdk.getSessionToken()
  40. </script>
  41. ''')
  42. await client.connected()
  43. token = await ui.run_javascript('return sessionToken && !sdk.isJwtExpired(sessionToken) ? sessionToken : null;')
  44. if token and self._verify(token):
  45. if self.path == '/login':
  46. await self.refresh_token()
  47. ui.open('/')
  48. else:
  49. func()
  50. else:
  51. if self.path != '/login':
  52. ui.open('/login')
  53. else:
  54. ui.timer(30, self.refresh_token)
  55. func()
  56. return super().__call__(content)
  57. @staticmethod
  58. def _verify(token: str) -> bool:
  59. try:
  60. descope_client.validate_session(session_token=token)
  61. return True
  62. except AuthException:
  63. logging.exception("Could not validate user session.")
  64. ui.notify('Wrong username or password', type='negative')
  65. return False
  66. @staticmethod
  67. async def refresh_token() -> None:
  68. await ui.run_javascript('sdk.refresh()', respond=False)
  69. def login_page(func: Callable[..., Any]) -> Callable[..., Any]:
  70. return page('/login')(func)