78 lines
3.1 KiB
Python
78 lines
3.1 KiB
Python
import secrets
|
||
import flet as ft
|
||
|
||
import api_client
|
||
|
||
_TOKEN_KEY = "auth_token"
|
||
|
||
|
||
class AuthService:
|
||
"""Сервис авторизации: логин, логаут, проверка токена.
|
||
HTTP-запросы делегируются в api_client.
|
||
"""
|
||
|
||
def __init__(self, page: ft.Page):
|
||
self._page = page
|
||
|
||
async def login(self, username: str, password: str) -> str | None:
|
||
"""POST /auth/login. Возвращает None при успехе, строку с ошибкой при неудаче."""
|
||
token, error = await api_client.login(username, password)
|
||
if error:
|
||
return error
|
||
self._page.session.store.set(_TOKEN_KEY, token)
|
||
self._page.session.store.set("url_key", secrets.token_hex(16)) # в адресную строку
|
||
self._page.session.store.set("session_key", secrets.token_hex(16)) # только в память
|
||
me = await api_client.get_me(token)
|
||
if me:
|
||
self._page.session.store.set("user_id", me["id"])
|
||
self._page.session.store.set("group_id", me.get("group_id") or "")
|
||
self._page.session.store.set("org_id", me.get("organization_id") or "")
|
||
return None
|
||
|
||
async def get_me(self) -> dict | None:
|
||
"""Возвращает данные текущего пользователя или None."""
|
||
token = self._page.session.store.get(_TOKEN_KEY)
|
||
if not token:
|
||
return None
|
||
return await api_client.get_me(token)
|
||
|
||
async def logout(self):
|
||
"""Инвалидирует токен на сервере и удаляет локальную сессию."""
|
||
token = self._page.session.store.get(_TOKEN_KEY)
|
||
if token:
|
||
await api_client.logout(token)
|
||
self._page.session.store.remove(_TOKEN_KEY)
|
||
self._page.session.store.remove("user_id")
|
||
self._page.session.store.remove("url_key")
|
||
self._page.session.store.remove("session_key")
|
||
self._page.session.store.remove("group_id")
|
||
self._page.session.store.remove("org_id")
|
||
|
||
async def get_token(self) -> str | None:
|
||
"""Возвращает сохранённый токен или None если не авторизован."""
|
||
return self._page.session.store.get(_TOKEN_KEY)
|
||
|
||
async def is_authenticated(self) -> bool:
|
||
token = await self.get_token()
|
||
return token is not None and token != ""
|
||
|
||
async def register(
|
||
self,
|
||
username: str,
|
||
password: str,
|
||
first_name: str,
|
||
last_name: str,
|
||
group_id: str | None = None,
|
||
organization_id: str | None = None,
|
||
) -> str | None:
|
||
"""POST /auth/register. Возвращает None при успехе, строку с ошибкой при неудаче."""
|
||
return await api_client.register(
|
||
username, password, first_name, last_name, group_id, organization_id
|
||
)
|
||
|
||
async def get_organizations(self) -> list:
|
||
return await api_client.get_organizations()
|
||
|
||
async def get_groups(self) -> list:
|
||
return await api_client.get_groups()
|