You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
126 lines
3.4 KiB
126 lines
3.4 KiB
"""Service to handle user_token operations."""
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Optional, Dict, Callable, Any
|
|
|
|
from atheneum.db import db
|
|
from atheneum.model import User, UserToken
|
|
from atheneum.service.serialization_service import (
|
|
BaseSerializer,
|
|
register_serializer
|
|
)
|
|
|
|
|
|
class UserTokenSerializer(BaseSerializer):
|
|
"""Serialize User model."""
|
|
|
|
def _serializers(self) -> Dict[str, Callable[[], Any]]:
|
|
"""Define the fields and the accompanying serializer factory."""
|
|
return {
|
|
'token': self.serialize_token,
|
|
'note': self.serialize_note,
|
|
'enabled': self.serialize_enabled,
|
|
'expirationTime': self.serialize_expiration_time,
|
|
'creationTime': self.serialize_creation_time,
|
|
'lastUsageTime': self.serialize_last_usage_time,
|
|
'version': self.serialize_version
|
|
}
|
|
|
|
def serialize_token(self) -> str:
|
|
"""User token."""
|
|
return self.model.token
|
|
|
|
def serialize_note(self) -> str:
|
|
"""User token note."""
|
|
return self.model.note
|
|
|
|
def serialize_enabled(self) -> bool:
|
|
"""User token enabled."""
|
|
return self.model.enabled
|
|
|
|
def serialize_expiration_time(self) -> datetime:
|
|
"""User token expiration time."""
|
|
return self.model.expiration_time
|
|
|
|
def serialize_creation_time(self) -> datetime:
|
|
"""User token creation time."""
|
|
return self.model.creation_time
|
|
|
|
def serialize_last_usage_time(self) -> datetime:
|
|
"""User token last usage time."""
|
|
return self.model.last_usage_time
|
|
|
|
def serialize_version(self) -> int:
|
|
"""User token version."""
|
|
return self.model.version
|
|
|
|
|
|
register_serializer(UserToken.__name__, UserTokenSerializer)
|
|
|
|
|
|
def generate_token() -> uuid.UUID:
|
|
"""
|
|
Generate a unique token.
|
|
|
|
:return:
|
|
"""
|
|
return uuid.uuid4()
|
|
|
|
|
|
def create(
|
|
user: User,
|
|
note: Optional[str] = None,
|
|
enabled: bool = True,
|
|
expiration_time: Optional[datetime] = None) -> UserToken:
|
|
"""
|
|
Create and save a UserToken.
|
|
|
|
:param user: The User object to bind the token to
|
|
:param note: An optional field to store additional information about a
|
|
token
|
|
:param enabled: A boolean to indicate whether a token can be considered
|
|
eligible for authentication
|
|
:param expiration_time: An optional argument to determine when the token
|
|
becomes invalid as a means of authentication. Defaults to None, which means
|
|
no expiration
|
|
:return:
|
|
"""
|
|
token = generate_token()
|
|
user_token = UserToken(
|
|
user_id=user.id,
|
|
token=token.__str__(),
|
|
note=note,
|
|
enabled=enabled,
|
|
creation_time=datetime.now(),
|
|
expiration_time=expiration_time,
|
|
version=0)
|
|
|
|
db.session.add(user_token)
|
|
db.session.commit()
|
|
|
|
return user_token
|
|
|
|
|
|
def delete(user_token: UserToken) -> bool:
|
|
"""
|
|
Delete a user_token.
|
|
|
|
:param user_token:
|
|
:return:
|
|
"""
|
|
existing_user_token = db.session.delete(user_token)
|
|
if existing_user_token is None:
|
|
db.session.commit()
|
|
return True
|
|
return False
|
|
|
|
|
|
def find_by_user_and_token(user: User, token: str) -> Optional[UserToken]:
|
|
"""
|
|
Lookup a user_token by user and token string.
|
|
|
|
:param user:
|
|
:param token:
|
|
:return:
|
|
"""
|
|
return UserToken.query.filter_by(user_id=user.id, token=token).first()
|