This module defines the Resonite client, which interacts with the Resonite API.

"""
This module defines the Resonite client, which interacts with the Resonite API.
"""

import threading
import concurrent.futures
import atexit
import asyncio
import re
import json
import os
import dataclasses
import logging
from datetime import datetime
from hashlib import sha256
from os import path
from typing import Dict, List, TypeVar, Literal
from importlib.resources import files

import requests
from requests.exceptions import JSONDecodeError as RequestsJSONDecodeError
from dateutil.parser import isoparse
from pydantic import TypeAdapter, ValidationError

from . import __version__
from .classes import (
    LoginDetails,
    ContactStatus,
    ResoniteDirectory,
    ResoniteContact,
    ResoniteLink,
    ResoniteRecord,
    ResoniteUser,
    ResoniteMessage,
    ResoniteCloudVar,
    recordTypeMapping,
    ResoniteSession,
    OwnerType,
    ResoniteCloudVarDefs,
    Platform,
    ResoniteUserMembership,
    ResoniteGroup,
    ResoniteGroupMember,
    ResoniteBadge,
)
from .utils import getOwnerType, deprecated_alias
from .endpoints import API_URL, ASSETS_URL
from resonitepy import exceptions as resonite_exceptions
from resonitepy.exceptions import (
    InvalidCredentials as ResoniteInvalidCredentials,
    InvalidToken as ResoniteInvalidToken,
    NoTokenError as ResoniteNoTokenError,
    ResoniteAPIException,
    ResoniteException,
)

T = TypeVar('T')

logger = logging.getLogger(__name__)

AUTHFILE_NAME = "auth.token"
# From PolyLogix/CloudX.js, the token seems to expire after 3600000 seconds (1 hour)
TOKEN_EXPIRY_SECONDS = 3600000

_TYPE_ADAPTERS: Dict[type, TypeAdapter] = {}

try:
    DEBUG = json.loads(os.environ.get('DEBUG', 'False').lower())
except json.decoder.JSONDecodeError:
    logger.error("Debug must be True or False")
    exit(1)

def to_class(data_class: type[T], data: dict) -> T:
    """ Converts a dictionary to an instance of the specified data class.

    Args:
        data_class (type): The type of the data class to convert to.
        data (dict): The dictionary containing the data to convert.

    Returns:
        object: An instance of the specified data class.

    Raises:
        ResoniteParseError: If the data doesn't match the class.

    Example:
        >>> data = {'globalVersion': 1, 'localVersion': 2}
        >>> record_version = to_class(ResoniteRecordVersion, data)
    """
    if data_class not in _TYPE_ADAPTERS:
        _TYPE_ADAPTERS[data_class] = TypeAdapter(data_class)
    try:
        return _TYPE_ADAPTERS[data_class].validate_python(data)
    except ValidationError as exc:
        logger.error('Error converting to class %s', data_class.__name__)
        if DEBUG:
            logger.error("With data: %s", data)
        raise resonite_exceptions.ResoniteParseError(data_class, data, exc) from exc


@dataclasses.dataclass
class Client:
    """ Representation of a Resonite client.

    This class provides methods for interacting with the Resonite API.

    Attributes:
        userId (str): The ID of the user associated with the client (in the Resonite `U-` format).
        token (str): The authentication token for the client.
        expire (str): The expiration date of the authentication token. (Not used by the API)
        rememberMe (bool): Whether to remember the client's session. (default: False)
        lastUpdate (datetime): The timestamp of the last update. (default: None)
        secretMachineIdHash (str): The hash of the secret machine ID. (default: None)
        secretMachineIdSalt (str): The salt of the secret machine ID. (default: None)
        session (Session): The session object used for making API requests. (default: Session())
        session.headers['UID'] (str): The UID header for the session. (default: randomly generated)

    Examples:
        >>> client = Client()
        >>> client.login(LoginDetails(username='foxxie', password='pass'))
        >>> inventory = client.getInventory()
    """
    userId: str = None
    token: str = None
    expire: datetime = None  # This don't seems to be use by the API.
    rememberMe: bool = False
    lastUpdate: datetime = None
    secretMachineIdHash: str = None
    secretMachineIdSalt: str = None
    session: requests.Session = None

    def __init__(self, cache: bool = True, cache_refresh_interval: float | None = 120):
        """Initialize a new Resonite API client instance."""
        self.session = requests.Session()
        self.session.headers['UID'] = sha256(os.urandom(16)).hexdigest().upper()
        self._hub = None
        self._hub_loop = None
        self._hub_thread = None
        self._hub_lock = threading.RLock()
        self._hub_cache = cache
        self._hub_cache_refresh_interval = cache_refresh_interval

    @property
    def headers(self) -> dict:
        """ Returns the headers for API requests.

        Returns:
            dict: A dictionary containing the headers.

        Examples:
            >>> client = Client()
            >>> headers = client.headers
        """
        default = {"User-Agent": f"resonitepy/{__version__}"}
        if not self.userId or not self.token:
            logger.warning("Headers not properly set. API requests might throw an error soon...")
            return default
        default["Authorization"] = f"res {self.userId}:{self.token}"
        return default

    def _start_hub(self):
        """ Start background event loop thread and connect the persistent hub.
        """
        from resonitepy.hub_manager import HubManager

        self._hub_loop = asyncio.new_event_loop()
        self._hub_thread = threading.Thread(
            target=self._hub_loop.run_forever, name="resonitepy-hub", daemon=True
        )
        self._hub_thread.start()

        hub = HubManager(
            self,
            cache = self._hub_cache,
            cache_refresh_interval = self._hub_cache_refresh_interval,
        )
        try:
            asyncio.run_coroutine_threadsafe(hub.connect(), self._hub_loop).result(timeout=15)
        except Exception:
            self.hub_close()
            raise
        self._hub = hub
        atexit.register(self.hub_close)

    def hub_close(self):
        """ Disconnect the persistent hub and stop its thread.
        """
        with self._hub_lock:
            if self._hub is not None and self._hub_loop is not None:
                try:
                    asyncio.run_coroutine_threadsafe(
                        self._hub.disconnect(), self._hub_loop
                    ).result(timeout=5)
                except Exception:
                    pass
            if self._hub_loop is not None:
                self._hub_loop.call_soon_threadsafe(self._hub_loop.stop)
                self._hub_thread.join(timeout=5)
                self._hub_loop.close()
            self._hub = None
            self._hub_loop = None
            self._hub_thread = None
            atexit.unregister(self.hub_close)

    def _hub_operation(self, operation):
        """ Run a one-shot hub operation from synchronous code.
        """
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            pass  # no loop running, safe to proceed
        else:
            raise resonite_exceptions.ResoniteHubException(
                "This method was called inside an async context - "
                "use HubManager directly (async) instead of the sync Client methods"
            )
        with self._hub_lock:
            if self._hub is not None and not self._hub.connected:
                self.hub_close()
            if self._hub is None:
                self._start_hub()
            hub, loop = self._hub, self._hub_loop
        future = asyncio.run_coroutine_threadsafe(operation(hub), loop)
        try:
            return future.result(timeout=30)
        except concurrent.futures.TimeoutError as e:
            raise resonite_exceptions.ResoniteHubException(
                "Hub operation timed out"
            ) from e

    def request(
            self,
            verb: str,
            path: str,
            data: dict = None,
            json: dict = None,
            params: dict = None,
            ignoreUpdate: bool = False
        ) -> Dict:
        """ Sends an API request and returns the response.

        While the API dont seems to implement more security, the official client
        behavior is respected. For now it will only disconnect after 1 day of
        inactivity.

        Args:
            verb (str): The HTTP verb for the request.
            path (str): The path of the API endpoint.
            data (str): The data to send in the request body. (default: None)
            json (dict): The JSON data to send in the request body. (default: None)
            params (dict): The query parameters for the request. (default: None)
            ignoreUpdate (bool): Whether to ignore the update check. (default: False)

        Returns:
            Dict: The response from the API.

        Raises:
            resonite_exceptions.InvalidCredentials: If the credentials are invalid.
            resonite_exceptions.InvalidToken: If the token is invalid.
            resonite_exceptions.ResoniteAPIException: If an API error occurs.

        Example:
            >>> client = Client()
            >>> response = client.request('get', '/users/U-foxxie')
        """

        # Check if session needs to be refreshed
        if self.lastUpdate and not ignoreUpdate:
            lastUpdate = self.lastUpdate
            if (datetime.now() - lastUpdate).total_seconds() <= TOKEN_EXPIRY_SECONDS:
                self.request('patch', '/userSessions', ignoreUpdate=True)
                self.lastUpdate = datetime.now()
            # TODO: Implement disconnection after 1 week of inactivity when implementing the rememberMe feature.
            #if 64800 >= (datetime.now() - lastUpdate).total_seconds() >= 85536:
            #    self.request('patch', '/userSessions', ignoreUpdate=True)
            else:
                raise resonite_exceptions.InvalidToken("Token expired")

        # Prepare request arguments
        args = {'url': API_URL + path}
        if data: args['data'] = data
        if json: args['json'] = json
        if params: args['params'] = params

        # Execute the request
        func = getattr(self.session, verb, None)

        with func(**args) as req:
            logger.debug("ResoniteAPI: [{}] {}".format(req.status_code, args))

            # Handle error responses
            if req.status_code not in [200, 204]:
                if "Invalid credentials" in req.text:
                    raise ResoniteInvalidCredentials(req.text)
                elif req.status_code == 403:
                    raise ResoniteInvalidToken(req.headers)
                else:
                    raise ResoniteAPIException(req)

            # Handle successful responses
            if req.status_code == 200:
                try:
                    response = req.json()
                    if "message" in response:
                        raise ResoniteAPIException(req, message=response["message"])
                    return response
                except RequestsJSONDecodeError:
                    return req.text

            # In case of a 204 response
            return

    def login(self, data: LoginDetails) -> None:
        """ Log in to the Resonite API with the provided login details.

        Args:
            data (LoginDetails): The login details.

        Returns:
            None

        Examples:
            >>> client = Client()
            >>> login_details = LoginDetails(username='foxxie', password='pass')
            >>> client.login(login_details)
        """
        payload = dataclasses.asdict(data)
        payload['authentication'] = data.authentication.build_dict()

        response = self.request('post', "/userSessions", json=payload)
        if not response:
            raise ResoniteException("Login failed - empty response")

        entity = response.get("entity", {})
        self.userId = entity.get("userId")
        self.token = entity.get("token")
        self.secretMachineIdHash = entity.get("secretMachineIdHash")
        self.secretMachineIdSalt = entity.get("secretMachineIdSalt")

        if "expire" in entity:
            self.expire = isoparse(entity["expire"])

        self.lastUpdate = datetime.now()
        self.session.headers.update(self.headers)

    def logout(self) -> None:
        """ Log out the current session.

        Returns:
            None

        Examples:
            >>> client = Client()
            >>> client.logout()
        """
        if self.userId and self.token:
            self.request(
                'delete',
                "/userSessions/{}/{}".format(self.userId, self.token),
                ignoreUpdate=True,
            )
        self.hub_close()
        self.clean_session()

    def clean_session(self) -> None:
        """ Clean the client's session datas.

        Returns:
            None

        Examples:
            >>> client = Client()
            >>> client.clean_session()
        """
        self.userId = None
        self.token = None
        self.expire = None
        self.secretMachineIdHash = None
        self.secretMachineIdSalt = None
        self.lastUpdate = None

        if "Authorization" in self.session.headers:
            del self.session.headers["Authorization"]
        self.session.headers.update(self.headers)

    def load_token(self) -> None:
        """ Load the authentication token from a file.

        Returns:
            None

        Raises:
            ResoniteNoTokenError: If the token file does not exist or the token has expired.

        Examples:
            >>> client = Client()
            >>> client.load_token()
        """
        if not path.exists(AUTHFILE_NAME):
            raise ResoniteNoTokenError("Auth token file not found.")

        with open(AUTHFILE_NAME, "r") as f:
            try:
                session = json.load(f)
                expire = datetime.fromisoformat(session.get("expire", ""))

                if datetime.now().timestamp() < expire.timestamp():
                    self.token = session.get("token")
                    self.userId = session.get("userId")
                    self.expire = expire
                    self.secretMachineIdHash = session.get("secretMachineIdHash")
                    self.secretMachineIdSalt = session.get("secretMachineIdSalt")
                    self.session.headers.update(self.headers)
                else:
                    raise ResoniteNoTokenError
            except (json.JSONDecodeError, ValueError):
                raise ResoniteNoTokenError("Invalid auth token file format.")

    def save_token(self) -> None:
        """ Saves the authentication token to a file.

        Returns:
            None

        Examples:
            >>> client = Client()
            >>> client.save_token()
        """
        if not all([self.userId, self.token, self.expire]):
            logger.warning("Cannot save token - missing required authentication data")
            return

        with open(AUTHFILE_NAME, "w+") as f:
            json.dump(
                {
                    "userId": self.userId,
                    "expire": self.expire.isoformat(),
                    "token": self.token,
                    "secretMachineIdHash": self.secretMachineIdHash,
                    "secretMachineIdSalt": self.secretMachineIdSalt,
                },
                f,
            )

    def res_db_signature(self, res_url: str) -> str:
        """ Returns the Resonite DB signature from a Resonite URL.

        Args:
            reres_url (url): The Resonite URL.

        Returns:
            str: The Resonite DB signature.

        Examples:
            >>> client = Client()
            >>> signature = client.res_db_signature("resrec://U-123/R-456")
        """
        parts = re.split("//+", res_url)
        if len(parts) > 2:
            raise ValueError(f"Invalid Resonite URL format: {res_url}")
        return parts[1].split(".")[0]

    @deprecated_alias(res_db_signature)
    def resDBSignature(self, resUrl: str) -> str:
        """ Returns the Resonite DB signature from a Resonite URL.

        Args:
            resUrl (url): The Resonite URL.

        Returns:
            str: The Resonite DB signature.

        Examples:
            >>> client = Client()
            >>> signature = client.resDBSignature("resrec://U-123/R-456")
        """
        return self.res_db_signature(res_url=resUrl)

    def res_db_to_http(self, res_url: str) -> str:
        """  Converts a Resonite URL to an HTTP URL.

        Args:
            res_url (str): The Resonite URL.

        Returns:
            str: The HTTP URL.

        # TODO: Fix example
        Examples:
            >>> client = Client()
            >>> http_url = client.res_db_signature("resrec://U-123/R-456")
        """
        return f"{ASSETS_URL.strip('/')}/{self.res_db_signature(res_url)}"

    @deprecated_alias(res_db_to_http)
    def resDbToHttp(self, resUrl: str) -> str:
        """  Converts a Resonite URL to an HTTP URL.

        Args:
            resUrl (str): The Resonite URL.

        Returns:
            str: The HTTP URL.

        Examples:
            >>> client = Client()
            >>> http_url = client.resDbToHttp("resrec://U-123/R-456")
        """
        return self.res_db_to_http(res_url=resUrl)

    @staticmethod
    def process_record_ist(data: List[dict]) -> List[ResoniteRecord]:
        """ Processes a list of raw records and returns a list of ResoniteRecord objects.

        Args:
            data: A list of dictionaries representing the raw records.

        Returns:
            A list of ResoniteRecord objects.
        """
        result = []
        for raw_item in data:
            item = to_class(ResoniteRecord, raw_item)
            record_class = recordTypeMapping.get(item.recordType, ResoniteRecord)
            x = to_class(record_class, raw_item)
            result.append(x)
        return result

    @staticmethod
    @deprecated_alias(process_record_ist)
    def processRecordList(data: list[dict]) -> List[ResoniteRecord]:
        """ Processes a list of raw records and returns a list of ResoniteRecord objects.

        Args:
            data: A list of dictionaries representing the raw records.

        Returns:
            A list of ResoniteRecord objects.
        """
        return Client.process_record_ist(data=data)

    def getUserData(self, user: str = None) -> ResoniteUser:
        """ Retrieves user data for the specified user.

        Args:
            user (str): The ID of the user to retrieve data for. If not provided, retrieves data for the client's user ID.

        Returns:
            ResoniteUser: A ResoniteUser object representing the user data.

        Examples:
            >>> client = Client()
            >>> user_data = client.getUserData()
        """
        if user is None:
            if not self.userId:
                logger.error("No user ID provided and client not logged in.")
            user = self.userId

        response = self.request('get', f"/users/{user}")
        return to_class(ResoniteUser, response)

    def getMemberships(self) -> List[ResoniteUserMembership]:
        """ Retrieve current connected user group memberships.

        Return
            List[ResoniteUserMembership]: The list of groups where the user is a member.

        """
        if not self.userId:
            logger.error("Client not logged in")

        response = self.request('get', f'/users/{self.userId}/Memberships')
        return [to_class(ResoniteUserMembership, group) for group in response]

    def getGroup(self, groupId: str) -> ResoniteGroup:
        """ Retrieve group information.

        Args:
            groupId (str): The group name starting with G-

        Returns:
            ResoniteGroup: An object with the group information.
        """
        response = self.request('get', f'/groups/{groupId}')
        return to_class(ResoniteGroup, response)

    def getGroupMembers(self, groupId: str) -> List[ResoniteGroupMember]:
        """ Retrieve members from a group.

        Args:
            groupId (str): The group name starting with G-

        Returns:
            List[ResoniteGroupMember]: A list with the group members.
        """
        response = self.request('get', f'/groups/{groupId}/members')
        return [to_class(ResoniteGroupMember, group_member) for group_member in response]

    def getGroupMember(self, groupId: str, userId: str) -> ResoniteGroupMember:
        """ Retrieve a member from a group.

        Args:
            groupId (str): The group name starting with G-
            userId (str): The username starting with U-
        Returns:
            ResoniteGroupMember: An object with the member information.
        """
        response = self.request('get', f'/groups/{groupId}/members/{userId}')
        return to_class(ResoniteGroupMember, response)

    def getSessions(
        self,
        compatibilityHash: str = None,
        name: str = None,
        universeId: str = None,
        hostName: str = None,
        hostId: str = None,
        minActiveUsers: int = 0,
        includeEmptyHeadless: bool = True,
    ) -> List[ResoniteSession]:
        """ Retrieves all active Resonite session.

        Args:
            compatibilityHash (str): The compatibility-hash sessions needs to have. A Resonite client can only join a session if this hash matches between client and server.
            name (str): The name of the session to search for.
            universeId (str): The id of the universe sessions need to be part of.
            hostName (str): The name of the user currently hosting the session.
            hostId (str): The id of the user currently hosting the session.
            minActiveUsers (int): he minimum amount of active users a session need to have. (default: 0)
            includeEmptyHeadless: (bool): Should empty headless servers be included in the results. (Default: True)

        Returns:
            List[ResoniteSession]: A list of ResoniteSessions

        Examples:
            >>> client = Client()
            >>> client.getSession()
        """
        # TODO: Implement the search for the sessions
        response = self.request('get', '/sessions')
        return [to_class(ResoniteSession, session) for session in response]

    def getSession(self, session_id: str) -> ResoniteSession:
        """ Retrieves session information for the specified session ID.

        Args:
            session_id (str): The ID of the session.

        Returns:
            ResoniteSession: A ResoniteSession object representing the session information.

        Examples:
            >>> client = Client()
            >>> session = client.getSession('12345')
        """
        response = self.request('get', f'/sessions/{session_id}')
        return to_class(ResoniteSession, response)

    def get_contacts(self, source: Literal["auto", "hub", "rest"] = "auto", force: bool = False) -> List[ResoniteContact]:
        """ Retrieves the contacts of the client.

        Returns:
            List[ResoniteContact]: A list of ResoniteContact objects representing the contacts.

        Examples:
            >>> client = Client()
            >>> contacts = client.getContacts()
        """
        if source == "auto":
            source = "hub" if (self._hub is not None and self._hub.connected) else "rest"
        if source == "hub":
            return self._hub_operation(lambda hub: hub.get_contacts(force=force))
        response = self.request('get', f"/users/{self.userId}/contacts")
        return [to_class(ResoniteContact, user) for user in response]

    @deprecated_alias(get_contacts)
    def getContacts(self) -> List[ResoniteContact]:
        """ Retrieves the contacts of the client.

        Returns:
            List[ResoniteContact]: A list of ResoniteContact objects representing the contacts.

        Examples:
            >>> client = Client()
            >>> contacts = client.getContacts()
        """
        return self.get_contacts()

    def add_contact(self, user_id: str):
        """ Send a contact request to a user.
        """
        self._hub_operation(lambda hub: hub.add_contact(user_id))

    def accept_contact_request(self, user_id: str):
        """ Accept a pending contact request.
        """
        self._hub_operation(lambda hub: hub.accept_contact_request(user_id))

    def decline_contact_request(self, user_id: str):
        """ Decline a pending contact request.
        """
        self._hub_operation(lambda hub: hub.decline_contact_request(user_id))

    def remove_contact(self, user_id: str):
        """ Remove a contact.
        """
        self._hub_operation(lambda hub: hub.remove_contact(user_id))

    def set_contact_status(self, user_id: str, status: ContactStatus, timeout: float = 5.0):
        """ Set user side of the relationship to an explicit status.
        See HubManager.set_contact_status for what each status does.
        """
        self._hub_operation(lambda hub: hub.set_contact_status(user_id, status, timeout))

    def getInventory(self) -> List[ResoniteRecord]:
        """ Retrieves the inventory of the user.

        Returns:
            List[ResoniteRecord]: A list of ResoniteRecord objects representing the inventory.

        Examples:
            >>> client = Client()
            >>> inventory = client.getInventory()
        """
        response = self.request(
            'get',
            f"/users/{self.userId}/records",
            params={"path": "Inventory"},
        )
        return self.processRecordList(response)

    def getDirectory(self, directory: ResoniteDirectory) -> List[ResoniteRecord]:
        """ Retrieves the contents of a directory.

        Args:
            directory (ResoniteDirectory): The ResoniteDirectory object representing the directory.

        Returns:
            List[ResoniteRecord]: A list of ResoniteRecord objects representing the contents of the directory.

        Examples:
            >>> client = Client()
            >>> directory = ResoniteDirectory(ownerId='U-123', content_path='path/to/directory')
            >>> contents = client.getDirectory(directory)
        """
        response = self.request(
            'get',
            f"/users/{directory.ownerId}/records",
            params={"path": directory.content_path},
        )
        return self.processRecordList(response)

    def resolveLink(self, link: ResoniteLink) -> ResoniteDirectory:
        """ Resolves a link type record and returns its directory.

        Args:
            link (ResoniteLink): The ResoniteLink object representing the link type record.

        Returns:
            ResoniteDirectory: A ResoniteDirectory object representing the directory.

        Raises:
            resonite_exceptions.ResoniteException: If the link type is not supported.
            resonite_exceptions.ResoniteAPIException: If the folder is not found in the cloud. Either delete or folder set back to non public. Supposed.
            resonite_exceptions.InvalidToken: If denied permission to access the folder in the cloud. Supposed.

        Examples:
            >>> client = Client()
            >>> link = ResoniteLink(assetUri=ParseResult(scheme='resrec', path='/G-Resonite/Inventory/Resonite Essentials'))  # This is not a valid ResoniteLink object but the minimal presend inside for this function to work
            >>> directory = client.resolveLink(link)
        """
        if link.assetUri.scheme != 'resrec':
            raise resonite_exceptions.ResoniteException(f"Not supported scheme '{link.assetUri.scheme}' for link type {link}")

        owner_id = None
        record = None
        record_type = None
        record_path = None

        match_user_link_legacy = re.search(r'\/(U-.*)\/(R-.*)', link.assetUri.path)
        if not record_path and match_user_link_legacy:
            owner_id = match_user_link_legacy.group(1)
            record = match_user_link_legacy.group(2)
            record_type = "users"
            record_path = (
                link.assetUri.path
                .replace('/'+owner_id+'/', '')
                .replace('/', '\\')
            )

        match_group_link_legacy = re.search(r'\/(G-.*)\/(R-.*)', link.assetUri.path)
        if not record_path and match_group_link_legacy:
            owner_id = match_group_link_legacy.group(1)
            record = match_group_link_legacy.group(2)
            record_type = "groups"
            record_path = (
                link.assetUri.path
                .replace('/'+owner_id+'/', '')
                .replace('/', '\\')
            )

        if link.id and link.assetUri.path:
            match_user_link = re.search(r'\/(U-.*?)\/', link.assetUri.path)
            if not record_path and match_user_link:
                owner_id = match_user_link.group(1)
                record_type = "users"
                record_path = (
                    link.assetUri.path
                    .replace('/'+owner_id+'/', '')
                    .replace('/', '\\')
                )
            match_group_link = re.search(r'\/(G-.*?)\/', link.assetUri.path)
            if not record_path and match_group_link:
                owner_id = match_group_link.group(1)
                record_type = "groups"
                record_path = (
                    link.assetUri.path
                    .replace('/'+owner_id+'/', '')
                    .replace('/', '\\')
                )
            record = link.id

        if not owner_id or not record or not record_type:
            raise resonite_exceptions.ResoniteException(f'Not supported group type in link type {link}')

        response = self.request(
            'get',
            f"/{record_type}/{owner_id}/records/{record_path}",
        )
        return to_class(ResoniteDirectory, response)

    def getMessageLegacy(
        self,
        fromTime: str = None,
        maxItems: int = 100,
        user: str = None,
        unreadOnly: bool = False
    ) -> List[ResoniteMessage]:
        """ Retrieves a list of Resonite messages.

        This API endpoint should be understand as deprecated. Please use the SignalR
        protocol for this instead.

        Args:
            fromTime (str): The starting time to retrieve messages from. (default: None, Not yet implemented)
            maxItems (int): The maximum number of messages to retrieve. (default: 100)
            user (str): The user ID to filter messages by. (default: None)
            unreadOnly (bool): Whether to retrieve only unread messages. (default: False)

        Returns:
            A list of ResoniteMessage objects.

        Raises:
            ValueError: If fromTime is provided (not yet implemented).

        Examples:
            >>> client = ResoniteClient()
            >>> messages = client.getMessageLegacy(maxItems=50, unreadOnly=True)
        """
        if fromTime:
            raise ValueError('fromTime parameter is not yet implemented')

        if not self.userId:
            logger.error("Client not logged in")

        params = {
            "maxItems": maxItems,
            "unreadOnly": unreadOnly,
        }

        if user:
            params['user'] = user

        response = self.request(
            'get',
            f'/users/{self.userId}/messages',
            params=params
        )

        messages = []
        for message in response:
            messages.append(to_class(ResoniteMessage, message))

        return messages

    def getOwnerPath(self, ownerId: str) -> str:
        """ Returns the owner path based on the owner ID.

        Args:
            ownerId (str): The ID of the owner.

        Returns:
            str: The owner path.

        Raises:
            ValueError: If the owner type is invalid.

        Examples:
            >>> client = Client()
            >>> owner_path = client.getOwnerPath('U-123')
        """
        ownerType = getOwnerType(ownerId)
        if ownerType == OwnerType.USER:
            return "users"
        elif ownerType == OwnerType.GROUP:
            return "groups"
        else:
            raise ValueError(f"invalid ownerType for {ownerId}")

    def listCloudVar(self, ownerId: str) -> List[ResoniteCloudVar]:
        """ Lists the cloud variables for the specified owner.

        Args:
            ownerId (str): The ID of the owner.

        Returns:
            List[ResoniteCloudVar]: A list of ResoniteCloudVar objects representing the cloud variables.

        Examples:
            >>> client = Client()
            >>> cloud_vars = client.listCloudVar('U-123')
        """
        response = self.request(
            'get',
            f'/{self.getOwnerPath(ownerId)}/{ownerId}/vars'
        )
        return [to_class(ResoniteCloudVar, cloud_var) for cloud_var in response]

    def getCloudVar(self, ownerId: str, path: str) -> ResoniteCloudVar:
        """ Retrieves a cloud variable for the specified owner and path.

        Args:
            ownerId (str): The ID of the owner.
            path (str): The path of the cloud variable.

        Returns:
            ResoniteCloudVar: A ResoniteCloudVar object representing the cloud variable.

        Examples:
            >>> client = Client()
            >>> cloud_var = client.getCloudVar('U-123', 'path/to/cloudvar')
        """
        response = self.request(
            'get',
            f'/{self.getOwnerPath(ownerId)}/{ownerId}/vars/{path}'
        )
        return to_class(ResoniteCloudVar, response)

    def getCloudVarDefs(self, ownerId: str, path: str) -> ResoniteCloudVarDefs:
        """ Retrieves the cloud variable definitions for the specified owner and path.

        Args:
            ownerId (str): The ID of the owner.
            path (str): The path of the cloud variable.

        Returns:
            ResoniteCloudVarDefs: A ResoniteCloudVarDefs object representing the cloud variable definitions.

        Raises:
            resonite_exceptions.ResoniteException: If the cloud variable doesn't exist.

        Examples:
            >>> client = Client()
            >>> defs = client.getCloudVarDefs('U-123', 'path/to/cloudvar')
        """
        json=[{
            "ownerId": ownerId,
            "path": path,
        }]

        response = self.request(
            'post',
            f'/readvars',
            json=json,
        )

        if not response:
            raise ResoniteException(f"{ownerId} {path} doesn't exist")

        return to_class(ResoniteCloudVarDefs, response[0]['definition'])

    def setCloudVar(self, ownerId: str, path: str, value: str) -> None:
        """ Sets the value of a cloud variable for the specified owner and path.

        Args:
            ownerId (str): The ID of the owner.
            path (str): The path of the cloud variable.
            value (str): The value to set for the cloud variable.

        Returns:
            None

        Examples:
            >>> client = Client()
            >>> client.setCloudVar('U-123', 'path/to/cloudvar', 'new value')
        """
        return self.request(
            'put',
            f'/{self.getOwnerPath(ownerId)}/{ownerId}/vars/{path}',
            json = {
                "ownerId": ownerId,
                "path": path,
                "value": value,
            }
        )

    def searchUser(self, username: str) -> List[ResoniteUser]:
        """ Searches for users based on username.

        This is not the U- Resonite user id, the API will search over usernames not ids.

        Args:
            username (str): The username to search for.

        Returns:
            List[ResoniteUser]: A list of ResoniteUser objects matching the search criteria.

        Examples:
            >>> client = Client()
            >>> users = client.searchUser('foxxie')
        """
        #TODO: the entitlements part could be optimized!
        response = self.request(
            'get',
            '/users',
            params = {'name': username}
        )
        users = []
        for user in response:
            users.append(to_class(ResoniteUser, user))
        return users

    def getUser(self, userId: str) -> ResoniteUser:
        """ Retrieve user directly.

        Args:
            userId (int): Ther username starting with U-

        Returns:
            ResoniteUser: The user
        """
        response = self.request('get', f'/users/{userId}')
        return to_class(ResoniteUser, response)

    def getUserByName(self, userId) -> ResoniteUser:
        reponse = self.request('get', f'/users/{userId}?byUsername=True')
        return to_class(ResoniteUser, reponse)

    def platform(self) -> Platform:
        """ Return information about the platform.
        """
        response = self.request('get', '/platform')
        return to_class(Platform, response)

    def badges(self) -> List[ResoniteBadge]:
        """ Return a list of ResoniteBadge object.

        For now this is a simple list of hadcoded badges from a CSV file.

        The badges with their tag surrounded `[]` are special badges which are not
        associated to a user directly but they are deduce from other data.

        - `[mobile]`: user's platform is set to Mobile
        - `[linux]`: user's platform is set to Linux
        - `[host]`: user is the session host
        - `[<year>]`: user's account was registred in <year> (<year> would be something like 2020 or 2025 depending)
        - `[baguette]`: user's account was registred at the same month and date as today but at a different year

        """
        badges = []
        path = files("resonitepy").joinpath("data/badges.csv")
        with path.open(encoding="utf-8", newline="") as f:
            rows = (line for line in f if not line.lstrip().startswith("#"))
            for row in rows:
                row = row.split(",")
                badge_dict = {
                    "tag": row[0],
                    "url": row[1],
                    "slotName": row[2]
                }
                badges.append(to_class(ResoniteBadge, badge_dict))
        return badges

Functions

def to_class(data_class: type[~T], data: dict) ‑> ~T

Converts a dictionary to an instance of the specified data class.

Args

-----=

data_class : type

The type of the data class to convert to.

data : dict

The dictionary containing the data to convert.

Returns

-----=

object

An instance of the specified data class.

Raises

-----=

ResoniteParseError

If the data doesn't match the class.

Example -----=

>>> data = {'globalVersion': 1, 'localVersion': 2}
>>> record_version = to_class(ResoniteRecordVersion, data)

Expand source code

def to_class(data_class: type[T], data: dict) -> T:
    """ Converts a dictionary to an instance of the specified data class.

    Args:
        data_class (type): The type of the data class to convert to.
        data (dict): The dictionary containing the data to convert.

    Returns:
        object: An instance of the specified data class.

    Raises:
        ResoniteParseError: If the data doesn't match the class.

    Example:
        >>> data = {'globalVersion': 1, 'localVersion': 2}
        >>> record_version = to_class(ResoniteRecordVersion, data)
    """
    if data_class not in _TYPE_ADAPTERS:
        _TYPE_ADAPTERS[data_class] = TypeAdapter(data_class)
    try:
        return _TYPE_ADAPTERS[data_class].validate_python(data)
    except ValidationError as exc:
        logger.error('Error converting to class %s', data_class.__name__)
        if DEBUG:
            logger.error("With data: %s", data)
        raise resonite_exceptions.ResoniteParseError(data_class, data, exc) from exc

Classes

class Client (cache: bool = True, cache_refresh_interval: float | None = 120)

Representation of a Resonite client.

This class provides methods for interacting with the Resonite API.

Attributes

-----=

userId : str

The ID of the user associated with the client (in the Resonite U- format).

token : str

The authentication token for the client.

expire : str

The expiration date of the authentication token. (Not used by the API)

rememberMe : bool

Whether to remember the client's session. (default: False)

lastUpdate : datetime

The timestamp of the last update. (default: None)

secretMachineIdHash : str

The hash of the secret machine ID. (default: None)

secretMachineIdSalt : str

The salt of the secret machine ID. (default: None)

session : Session

The session object used for making API requests. (default: Session())

session.headers['UID'] (str): The UID header for the session. (default: randomly generated)

Examples -----=

>>> client = Client()
>>> client.login(LoginDetails(username='foxxie', password='pass'))
>>> inventory = client.getInventory()

Initialize a new Resonite API client instance.

Expand source code

@dataclasses.dataclass
class Client:
    """ Representation of a Resonite client.

    This class provides methods for interacting with the Resonite API.

    Attributes:
        userId (str): The ID of the user associated with the client (in the Resonite `U-` format).
        token (str): The authentication token for the client.
        expire (str): The expiration date of the authentication token. (Not used by the API)
        rememberMe (bool): Whether to remember the client's session. (default: False)
        lastUpdate (datetime): The timestamp of the last update. (default: None)
        secretMachineIdHash (str): The hash of the secret machine ID. (default: None)
        secretMachineIdSalt (str): The salt of the secret machine ID. (default: None)
        session (Session): The session object used for making API requests. (default: Session())
        session.headers['UID'] (str): The UID header for the session. (default: randomly generated)

    Examples:
        >>> client = Client()
        >>> client.login(LoginDetails(username='foxxie', password='pass'))
        >>> inventory = client.getInventory()
    """
    userId: str = None
    token: str = None
    expire: datetime = None  # This don't seems to be use by the API.
    rememberMe: bool = False
    lastUpdate: datetime = None
    secretMachineIdHash: str = None
    secretMachineIdSalt: str = None
    session: requests.Session = None

    def __init__(self, cache: bool = True, cache_refresh_interval: float | None = 120):
        """Initialize a new Resonite API client instance."""
        self.session = requests.Session()
        self.session.headers['UID'] = sha256(os.urandom(16)).hexdigest().upper()
        self._hub = None
        self._hub_loop = None
        self._hub_thread = None
        self._hub_lock = threading.RLock()
        self._hub_cache = cache
        self._hub_cache_refresh_interval = cache_refresh_interval

    @property
    def headers(self) -> dict:
        """ Returns the headers for API requests.

        Returns:
            dict: A dictionary containing the headers.

        Examples:
            >>> client = Client()
            >>> headers = client.headers
        """
        default = {"User-Agent": f"resonitepy/{__version__}"}
        if not self.userId or not self.token:
            logger.warning("Headers not properly set. API requests might throw an error soon...")
            return default
        default["Authorization"] = f"res {self.userId}:{self.token}"
        return default

    def _start_hub(self):
        """ Start background event loop thread and connect the persistent hub.
        """
        from resonitepy.hub_manager import HubManager

        self._hub_loop = asyncio.new_event_loop()
        self._hub_thread = threading.Thread(
            target=self._hub_loop.run_forever, name="resonitepy-hub", daemon=True
        )
        self._hub_thread.start()

        hub = HubManager(
            self,
            cache = self._hub_cache,
            cache_refresh_interval = self._hub_cache_refresh_interval,
        )
        try:
            asyncio.run_coroutine_threadsafe(hub.connect(), self._hub_loop).result(timeout=15)
        except Exception:
            self.hub_close()
            raise
        self._hub = hub
        atexit.register(self.hub_close)

    def hub_close(self):
        """ Disconnect the persistent hub and stop its thread.
        """
        with self._hub_lock:
            if self._hub is not None and self._hub_loop is not None:
                try:
                    asyncio.run_coroutine_threadsafe(
                        self._hub.disconnect(), self._hub_loop
                    ).result(timeout=5)
                except Exception:
                    pass
            if self._hub_loop is not None:
                self._hub_loop.call_soon_threadsafe(self._hub_loop.stop)
                self._hub_thread.join(timeout=5)
                self._hub_loop.close()
            self._hub = None
            self._hub_loop = None
            self._hub_thread = None
            atexit.unregister(self.hub_close)

    def _hub_operation(self, operation):
        """ Run a one-shot hub operation from synchronous code.
        """
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            pass  # no loop running, safe to proceed
        else:
            raise resonite_exceptions.ResoniteHubException(
                "This method was called inside an async context - "
                "use HubManager directly (async) instead of the sync Client methods"
            )
        with self._hub_lock:
            if self._hub is not None and not self._hub.connected:
                self.hub_close()
            if self._hub is None:
                self._start_hub()
            hub, loop = self._hub, self._hub_loop
        future = asyncio.run_coroutine_threadsafe(operation(hub), loop)
        try:
            return future.result(timeout=30)
        except concurrent.futures.TimeoutError as e:
            raise resonite_exceptions.ResoniteHubException(
                "Hub operation timed out"
            ) from e

    def request(
            self,
            verb: str,
            path: str,
            data: dict = None,
            json: dict = None,
            params: dict = None,
            ignoreUpdate: bool = False
        ) -> Dict:
        """ Sends an API request and returns the response.

        While the API dont seems to implement more security, the official client
        behavior is respected. For now it will only disconnect after 1 day of
        inactivity.

        Args:
            verb (str): The HTTP verb for the request.
            path (str): The path of the API endpoint.
            data (str): The data to send in the request body. (default: None)
            json (dict): The JSON data to send in the request body. (default: None)
            params (dict): The query parameters for the request. (default: None)
            ignoreUpdate (bool): Whether to ignore the update check. (default: False)

        Returns:
            Dict: The response from the API.

        Raises:
            resonite_exceptions.InvalidCredentials: If the credentials are invalid.
            resonite_exceptions.InvalidToken: If the token is invalid.
            resonite_exceptions.ResoniteAPIException: If an API error occurs.

        Example:
            >>> client = Client()
            >>> response = client.request('get', '/users/U-foxxie')
        """

        # Check if session needs to be refreshed
        if self.lastUpdate and not ignoreUpdate:
            lastUpdate = self.lastUpdate
            if (datetime.now() - lastUpdate).total_seconds() <= TOKEN_EXPIRY_SECONDS:
                self.request('patch', '/userSessions', ignoreUpdate=True)
                self.lastUpdate = datetime.now()
            # TODO: Implement disconnection after 1 week of inactivity when implementing the rememberMe feature.
            #if 64800 >= (datetime.now() - lastUpdate).total_seconds() >= 85536:
            #    self.request('patch', '/userSessions', ignoreUpdate=True)
            else:
                raise resonite_exceptions.InvalidToken("Token expired")

        # Prepare request arguments
        args = {'url': API_URL + path}
        if data: args['data'] = data
        if json: args['json'] = json
        if params: args['params'] = params

        # Execute the request
        func = getattr(self.session, verb, None)

        with func(**args) as req:
            logger.debug("ResoniteAPI: [{}] {}".format(req.status_code, args))

            # Handle error responses
            if req.status_code not in [200, 204]:
                if "Invalid credentials" in req.text:
                    raise ResoniteInvalidCredentials(req.text)
                elif req.status_code == 403:
                    raise ResoniteInvalidToken(req.headers)
                else:
                    raise ResoniteAPIException(req)

            # Handle successful responses
            if req.status_code == 200:
                try:
                    response = req.json()
                    if "message" in response:
                        raise ResoniteAPIException(req, message=response["message"])
                    return response
                except RequestsJSONDecodeError:
                    return req.text

            # In case of a 204 response
            return

    def login(self, data: LoginDetails) -> None:
        """ Log in to the Resonite API with the provided login details.

        Args:
            data (LoginDetails): The login details.

        Returns:
            None

        Examples:
            >>> client = Client()
            >>> login_details = LoginDetails(username='foxxie', password='pass')
            >>> client.login(login_details)
        """
        payload = dataclasses.asdict(data)
        payload['authentication'] = data.authentication.build_dict()

        response = self.request('post', "/userSessions", json=payload)
        if not response:
            raise ResoniteException("Login failed - empty response")

        entity = response.get("entity", {})
        self.userId = entity.get("userId")
        self.token = entity.get("token")
        self.secretMachineIdHash = entity.get("secretMachineIdHash")
        self.secretMachineIdSalt = entity.get("secretMachineIdSalt")

        if "expire" in entity:
            self.expire = isoparse(entity["expire"])

        self.lastUpdate = datetime.now()
        self.session.headers.update(self.headers)

    def logout(self) -> None:
        """ Log out the current session.

        Returns:
            None

        Examples:
            >>> client = Client()
            >>> client.logout()
        """
        if self.userId and self.token:
            self.request(
                'delete',
                "/userSessions/{}/{}".format(self.userId, self.token),
                ignoreUpdate=True,
            )
        self.hub_close()
        self.clean_session()

    def clean_session(self) -> None:
        """ Clean the client's session datas.

        Returns:
            None

        Examples:
            >>> client = Client()
            >>> client.clean_session()
        """
        self.userId = None
        self.token = None
        self.expire = None
        self.secretMachineIdHash = None
        self.secretMachineIdSalt = None
        self.lastUpdate = None

        if "Authorization" in self.session.headers:
            del self.session.headers["Authorization"]
        self.session.headers.update(self.headers)

    def load_token(self) -> None:
        """ Load the authentication token from a file.

        Returns:
            None

        Raises:
            ResoniteNoTokenError: If the token file does not exist or the token has expired.

        Examples:
            >>> client = Client()
            >>> client.load_token()
        """
        if not path.exists(AUTHFILE_NAME):
            raise ResoniteNoTokenError("Auth token file not found.")

        with open(AUTHFILE_NAME, "r") as f:
            try:
                session = json.load(f)
                expire = datetime.fromisoformat(session.get("expire", ""))

                if datetime.now().timestamp() < expire.timestamp():
                    self.token = session.get("token")
                    self.userId = session.get("userId")
                    self.expire = expire
                    self.secretMachineIdHash = session.get("secretMachineIdHash")
                    self.secretMachineIdSalt = session.get("secretMachineIdSalt")
                    self.session.headers.update(self.headers)
                else:
                    raise ResoniteNoTokenError
            except (json.JSONDecodeError, ValueError):
                raise ResoniteNoTokenError("Invalid auth token file format.")

    def save_token(self) -> None:
        """ Saves the authentication token to a file.

        Returns:
            None

        Examples:
            >>> client = Client()
            >>> client.save_token()
        """
        if not all([self.userId, self.token, self.expire]):
            logger.warning("Cannot save token - missing required authentication data")
            return

        with open(AUTHFILE_NAME, "w+") as f:
            json.dump(
                {
                    "userId": self.userId,
                    "expire": self.expire.isoformat(),
                    "token": self.token,
                    "secretMachineIdHash": self.secretMachineIdHash,
                    "secretMachineIdSalt": self.secretMachineIdSalt,
                },
                f,
            )

    def res_db_signature(self, res_url: str) -> str:
        """ Returns the Resonite DB signature from a Resonite URL.

        Args:
            reres_url (url): The Resonite URL.

        Returns:
            str: The Resonite DB signature.

        Examples:
            >>> client = Client()
            >>> signature = client.res_db_signature("resrec://U-123/R-456")
        """
        parts = re.split("//+", res_url)
        if len(parts) > 2:
            raise ValueError(f"Invalid Resonite URL format: {res_url}")
        return parts[1].split(".")[0]

    @deprecated_alias(res_db_signature)
    def resDBSignature(self, resUrl: str) -> str:
        """ Returns the Resonite DB signature from a Resonite URL.

        Args:
            resUrl (url): The Resonite URL.

        Returns:
            str: The Resonite DB signature.

        Examples:
            >>> client = Client()
            >>> signature = client.resDBSignature("resrec://U-123/R-456")
        """
        return self.res_db_signature(res_url=resUrl)

    def res_db_to_http(self, res_url: str) -> str:
        """  Converts a Resonite URL to an HTTP URL.

        Args:
            res_url (str): The Resonite URL.

        Returns:
            str: The HTTP URL.

        # TODO: Fix example
        Examples:
            >>> client = Client()
            >>> http_url = client.res_db_signature("resrec://U-123/R-456")
        """
        return f"{ASSETS_URL.strip('/')}/{self.res_db_signature(res_url)}"

    @deprecated_alias(res_db_to_http)
    def resDbToHttp(self, resUrl: str) -> str:
        """  Converts a Resonite URL to an HTTP URL.

        Args:
            resUrl (str): The Resonite URL.

        Returns:
            str: The HTTP URL.

        Examples:
            >>> client = Client()
            >>> http_url = client.resDbToHttp("resrec://U-123/R-456")
        """
        return self.res_db_to_http(res_url=resUrl)

    @staticmethod
    def process_record_ist(data: List[dict]) -> List[ResoniteRecord]:
        """ Processes a list of raw records and returns a list of ResoniteRecord objects.

        Args:
            data: A list of dictionaries representing the raw records.

        Returns:
            A list of ResoniteRecord objects.
        """
        result = []
        for raw_item in data:
            item = to_class(ResoniteRecord, raw_item)
            record_class = recordTypeMapping.get(item.recordType, ResoniteRecord)
            x = to_class(record_class, raw_item)
            result.append(x)
        return result

    @staticmethod
    @deprecated_alias(process_record_ist)
    def processRecordList(data: list[dict]) -> List[ResoniteRecord]:
        """ Processes a list of raw records and returns a list of ResoniteRecord objects.

        Args:
            data: A list of dictionaries representing the raw records.

        Returns:
            A list of ResoniteRecord objects.
        """
        return Client.process_record_ist(data=data)

    def getUserData(self, user: str = None) -> ResoniteUser:
        """ Retrieves user data for the specified user.

        Args:
            user (str): The ID of the user to retrieve data for. If not provided, retrieves data for the client's user ID.

        Returns:
            ResoniteUser: A ResoniteUser object representing the user data.

        Examples:
            >>> client = Client()
            >>> user_data = client.getUserData()
        """
        if user is None:
            if not self.userId:
                logger.error("No user ID provided and client not logged in.")
            user = self.userId

        response = self.request('get', f"/users/{user}")
        return to_class(ResoniteUser, response)

    def getMemberships(self) -> List[ResoniteUserMembership]:
        """ Retrieve current connected user group memberships.

        Return
            List[ResoniteUserMembership]: The list of groups where the user is a member.

        """
        if not self.userId:
            logger.error("Client not logged in")

        response = self.request('get', f'/users/{self.userId}/Memberships')
        return [to_class(ResoniteUserMembership, group) for group in response]

    def getGroup(self, groupId: str) -> ResoniteGroup:
        """ Retrieve group information.

        Args:
            groupId (str): The group name starting with G-

        Returns:
            ResoniteGroup: An object with the group information.
        """
        response = self.request('get', f'/groups/{groupId}')
        return to_class(ResoniteGroup, response)

    def getGroupMembers(self, groupId: str) -> List[ResoniteGroupMember]:
        """ Retrieve members from a group.

        Args:
            groupId (str): The group name starting with G-

        Returns:
            List[ResoniteGroupMember]: A list with the group members.
        """
        response = self.request('get', f'/groups/{groupId}/members')
        return [to_class(ResoniteGroupMember, group_member) for group_member in response]

    def getGroupMember(self, groupId: str, userId: str) -> ResoniteGroupMember:
        """ Retrieve a member from a group.

        Args:
            groupId (str): The group name starting with G-
            userId (str): The username starting with U-
        Returns:
            ResoniteGroupMember: An object with the member information.
        """
        response = self.request('get', f'/groups/{groupId}/members/{userId}')
        return to_class(ResoniteGroupMember, response)

    def getSessions(
        self,
        compatibilityHash: str = None,
        name: str = None,
        universeId: str = None,
        hostName: str = None,
        hostId: str = None,
        minActiveUsers: int = 0,
        includeEmptyHeadless: bool = True,
    ) -> List[ResoniteSession]:
        """ Retrieves all active Resonite session.

        Args:
            compatibilityHash (str): The compatibility-hash sessions needs to have. A Resonite client can only join a session if this hash matches between client and server.
            name (str): The name of the session to search for.
            universeId (str): The id of the universe sessions need to be part of.
            hostName (str): The name of the user currently hosting the session.
            hostId (str): The id of the user currently hosting the session.
            minActiveUsers (int): he minimum amount of active users a session need to have. (default: 0)
            includeEmptyHeadless: (bool): Should empty headless servers be included in the results. (Default: True)

        Returns:
            List[ResoniteSession]: A list of ResoniteSessions

        Examples:
            >>> client = Client()
            >>> client.getSession()
        """
        # TODO: Implement the search for the sessions
        response = self.request('get', '/sessions')
        return [to_class(ResoniteSession, session) for session in response]

    def getSession(self, session_id: str) -> ResoniteSession:
        """ Retrieves session information for the specified session ID.

        Args:
            session_id (str): The ID of the session.

        Returns:
            ResoniteSession: A ResoniteSession object representing the session information.

        Examples:
            >>> client = Client()
            >>> session = client.getSession('12345')
        """
        response = self.request('get', f'/sessions/{session_id}')
        return to_class(ResoniteSession, response)

    def get_contacts(self, source: Literal["auto", "hub", "rest"] = "auto", force: bool = False) -> List[ResoniteContact]:
        """ Retrieves the contacts of the client.

        Returns:
            List[ResoniteContact]: A list of ResoniteContact objects representing the contacts.

        Examples:
            >>> client = Client()
            >>> contacts = client.getContacts()
        """
        if source == "auto":
            source = "hub" if (self._hub is not None and self._hub.connected) else "rest"
        if source == "hub":
            return self._hub_operation(lambda hub: hub.get_contacts(force=force))
        response = self.request('get', f"/users/{self.userId}/contacts")
        return [to_class(ResoniteContact, user) for user in response]

    @deprecated_alias(get_contacts)
    def getContacts(self) -> List[ResoniteContact]:
        """ Retrieves the contacts of the client.

        Returns:
            List[ResoniteContact]: A list of ResoniteContact objects representing the contacts.

        Examples:
            >>> client = Client()
            >>> contacts = client.getContacts()
        """
        return self.get_contacts()

    def add_contact(self, user_id: str):
        """ Send a contact request to a user.
        """
        self._hub_operation(lambda hub: hub.add_contact(user_id))

    def accept_contact_request(self, user_id: str):
        """ Accept a pending contact request.
        """
        self._hub_operation(lambda hub: hub.accept_contact_request(user_id))

    def decline_contact_request(self, user_id: str):
        """ Decline a pending contact request.
        """
        self._hub_operation(lambda hub: hub.decline_contact_request(user_id))

    def remove_contact(self, user_id: str):
        """ Remove a contact.
        """
        self._hub_operation(lambda hub: hub.remove_contact(user_id))

    def set_contact_status(self, user_id: str, status: ContactStatus, timeout: float = 5.0):
        """ Set user side of the relationship to an explicit status.
        See HubManager.set_contact_status for what each status does.
        """
        self._hub_operation(lambda hub: hub.set_contact_status(user_id, status, timeout))

    def getInventory(self) -> List[ResoniteRecord]:
        """ Retrieves the inventory of the user.

        Returns:
            List[ResoniteRecord]: A list of ResoniteRecord objects representing the inventory.

        Examples:
            >>> client = Client()
            >>> inventory = client.getInventory()
        """
        response = self.request(
            'get',
            f"/users/{self.userId}/records",
            params={"path": "Inventory"},
        )
        return self.processRecordList(response)

    def getDirectory(self, directory: ResoniteDirectory) -> List[ResoniteRecord]:
        """ Retrieves the contents of a directory.

        Args:
            directory (ResoniteDirectory): The ResoniteDirectory object representing the directory.

        Returns:
            List[ResoniteRecord]: A list of ResoniteRecord objects representing the contents of the directory.

        Examples:
            >>> client = Client()
            >>> directory = ResoniteDirectory(ownerId='U-123', content_path='path/to/directory')
            >>> contents = client.getDirectory(directory)
        """
        response = self.request(
            'get',
            f"/users/{directory.ownerId}/records",
            params={"path": directory.content_path},
        )
        return self.processRecordList(response)

    def resolveLink(self, link: ResoniteLink) -> ResoniteDirectory:
        """ Resolves a link type record and returns its directory.

        Args:
            link (ResoniteLink): The ResoniteLink object representing the link type record.

        Returns:
            ResoniteDirectory: A ResoniteDirectory object representing the directory.

        Raises:
            resonite_exceptions.ResoniteException: If the link type is not supported.
            resonite_exceptions.ResoniteAPIException: If the folder is not found in the cloud. Either delete or folder set back to non public. Supposed.
            resonite_exceptions.InvalidToken: If denied permission to access the folder in the cloud. Supposed.

        Examples:
            >>> client = Client()
            >>> link = ResoniteLink(assetUri=ParseResult(scheme='resrec', path='/G-Resonite/Inventory/Resonite Essentials'))  # This is not a valid ResoniteLink object but the minimal presend inside for this function to work
            >>> directory = client.resolveLink(link)
        """
        if link.assetUri.scheme != 'resrec':
            raise resonite_exceptions.ResoniteException(f"Not supported scheme '{link.assetUri.scheme}' for link type {link}")

        owner_id = None
        record = None
        record_type = None
        record_path = None

        match_user_link_legacy = re.search(r'\/(U-.*)\/(R-.*)', link.assetUri.path)
        if not record_path and match_user_link_legacy:
            owner_id = match_user_link_legacy.group(1)
            record = match_user_link_legacy.group(2)
            record_type = "users"
            record_path = (
                link.assetUri.path
                .replace('/'+owner_id+'/', '')
                .replace('/', '\\')
            )

        match_group_link_legacy = re.search(r'\/(G-.*)\/(R-.*)', link.assetUri.path)
        if not record_path and match_group_link_legacy:
            owner_id = match_group_link_legacy.group(1)
            record = match_group_link_legacy.group(2)
            record_type = "groups"
            record_path = (
                link.assetUri.path
                .replace('/'+owner_id+'/', '')
                .replace('/', '\\')
            )

        if link.id and link.assetUri.path:
            match_user_link = re.search(r'\/(U-.*?)\/', link.assetUri.path)
            if not record_path and match_user_link:
                owner_id = match_user_link.group(1)
                record_type = "users"
                record_path = (
                    link.assetUri.path
                    .replace('/'+owner_id+'/', '')
                    .replace('/', '\\')
                )
            match_group_link = re.search(r'\/(G-.*?)\/', link.assetUri.path)
            if not record_path and match_group_link:
                owner_id = match_group_link.group(1)
                record_type = "groups"
                record_path = (
                    link.assetUri.path
                    .replace('/'+owner_id+'/', '')
                    .replace('/', '\\')
                )
            record = link.id

        if not owner_id or not record or not record_type:
            raise resonite_exceptions.ResoniteException(f'Not supported group type in link type {link}')

        response = self.request(
            'get',
            f"/{record_type}/{owner_id}/records/{record_path}",
        )
        return to_class(ResoniteDirectory, response)

    def getMessageLegacy(
        self,
        fromTime: str = None,
        maxItems: int = 100,
        user: str = None,
        unreadOnly: bool = False
    ) -> List[ResoniteMessage]:
        """ Retrieves a list of Resonite messages.

        This API endpoint should be understand as deprecated. Please use the SignalR
        protocol for this instead.

        Args:
            fromTime (str): The starting time to retrieve messages from. (default: None, Not yet implemented)
            maxItems (int): The maximum number of messages to retrieve. (default: 100)
            user (str): The user ID to filter messages by. (default: None)
            unreadOnly (bool): Whether to retrieve only unread messages. (default: False)

        Returns:
            A list of ResoniteMessage objects.

        Raises:
            ValueError: If fromTime is provided (not yet implemented).

        Examples:
            >>> client = ResoniteClient()
            >>> messages = client.getMessageLegacy(maxItems=50, unreadOnly=True)
        """
        if fromTime:
            raise ValueError('fromTime parameter is not yet implemented')

        if not self.userId:
            logger.error("Client not logged in")

        params = {
            "maxItems": maxItems,
            "unreadOnly": unreadOnly,
        }

        if user:
            params['user'] = user

        response = self.request(
            'get',
            f'/users/{self.userId}/messages',
            params=params
        )

        messages = []
        for message in response:
            messages.append(to_class(ResoniteMessage, message))

        return messages

    def getOwnerPath(self, ownerId: str) -> str:
        """ Returns the owner path based on the owner ID.

        Args:
            ownerId (str): The ID of the owner.

        Returns:
            str: The owner path.

        Raises:
            ValueError: If the owner type is invalid.

        Examples:
            >>> client = Client()
            >>> owner_path = client.getOwnerPath('U-123')
        """
        ownerType = getOwnerType(ownerId)
        if ownerType == OwnerType.USER:
            return "users"
        elif ownerType == OwnerType.GROUP:
            return "groups"
        else:
            raise ValueError(f"invalid ownerType for {ownerId}")

    def listCloudVar(self, ownerId: str) -> List[ResoniteCloudVar]:
        """ Lists the cloud variables for the specified owner.

        Args:
            ownerId (str): The ID of the owner.

        Returns:
            List[ResoniteCloudVar]: A list of ResoniteCloudVar objects representing the cloud variables.

        Examples:
            >>> client = Client()
            >>> cloud_vars = client.listCloudVar('U-123')
        """
        response = self.request(
            'get',
            f'/{self.getOwnerPath(ownerId)}/{ownerId}/vars'
        )
        return [to_class(ResoniteCloudVar, cloud_var) for cloud_var in response]

    def getCloudVar(self, ownerId: str, path: str) -> ResoniteCloudVar:
        """ Retrieves a cloud variable for the specified owner and path.

        Args:
            ownerId (str): The ID of the owner.
            path (str): The path of the cloud variable.

        Returns:
            ResoniteCloudVar: A ResoniteCloudVar object representing the cloud variable.

        Examples:
            >>> client = Client()
            >>> cloud_var = client.getCloudVar('U-123', 'path/to/cloudvar')
        """
        response = self.request(
            'get',
            f'/{self.getOwnerPath(ownerId)}/{ownerId}/vars/{path}'
        )
        return to_class(ResoniteCloudVar, response)

    def getCloudVarDefs(self, ownerId: str, path: str) -> ResoniteCloudVarDefs:
        """ Retrieves the cloud variable definitions for the specified owner and path.

        Args:
            ownerId (str): The ID of the owner.
            path (str): The path of the cloud variable.

        Returns:
            ResoniteCloudVarDefs: A ResoniteCloudVarDefs object representing the cloud variable definitions.

        Raises:
            resonite_exceptions.ResoniteException: If the cloud variable doesn't exist.

        Examples:
            >>> client = Client()
            >>> defs = client.getCloudVarDefs('U-123', 'path/to/cloudvar')
        """
        json=[{
            "ownerId": ownerId,
            "path": path,
        }]

        response = self.request(
            'post',
            f'/readvars',
            json=json,
        )

        if not response:
            raise ResoniteException(f"{ownerId} {path} doesn't exist")

        return to_class(ResoniteCloudVarDefs, response[0]['definition'])

    def setCloudVar(self, ownerId: str, path: str, value: str) -> None:
        """ Sets the value of a cloud variable for the specified owner and path.

        Args:
            ownerId (str): The ID of the owner.
            path (str): The path of the cloud variable.
            value (str): The value to set for the cloud variable.

        Returns:
            None

        Examples:
            >>> client = Client()
            >>> client.setCloudVar('U-123', 'path/to/cloudvar', 'new value')
        """
        return self.request(
            'put',
            f'/{self.getOwnerPath(ownerId)}/{ownerId}/vars/{path}',
            json = {
                "ownerId": ownerId,
                "path": path,
                "value": value,
            }
        )

    def searchUser(self, username: str) -> List[ResoniteUser]:
        """ Searches for users based on username.

        This is not the U- Resonite user id, the API will search over usernames not ids.

        Args:
            username (str): The username to search for.

        Returns:
            List[ResoniteUser]: A list of ResoniteUser objects matching the search criteria.

        Examples:
            >>> client = Client()
            >>> users = client.searchUser('foxxie')
        """
        #TODO: the entitlements part could be optimized!
        response = self.request(
            'get',
            '/users',
            params = {'name': username}
        )
        users = []
        for user in response:
            users.append(to_class(ResoniteUser, user))
        return users

    def getUser(self, userId: str) -> ResoniteUser:
        """ Retrieve user directly.

        Args:
            userId (int): Ther username starting with U-

        Returns:
            ResoniteUser: The user
        """
        response = self.request('get', f'/users/{userId}')
        return to_class(ResoniteUser, response)

    def getUserByName(self, userId) -> ResoniteUser:
        reponse = self.request('get', f'/users/{userId}?byUsername=True')
        return to_class(ResoniteUser, reponse)

    def platform(self) -> Platform:
        """ Return information about the platform.
        """
        response = self.request('get', '/platform')
        return to_class(Platform, response)

    def badges(self) -> List[ResoniteBadge]:
        """ Return a list of ResoniteBadge object.

        For now this is a simple list of hadcoded badges from a CSV file.

        The badges with their tag surrounded `[]` are special badges which are not
        associated to a user directly but they are deduce from other data.

        - `[mobile]`: user's platform is set to Mobile
        - `[linux]`: user's platform is set to Linux
        - `[host]`: user is the session host
        - `[<year>]`: user's account was registred in <year> (<year> would be something like 2020 or 2025 depending)
        - `[baguette]`: user's account was registred at the same month and date as today but at a different year

        """
        badges = []
        path = files("resonitepy").joinpath("data/badges.csv")
        with path.open(encoding="utf-8", newline="") as f:
            rows = (line for line in f if not line.lstrip().startswith("#"))
            for row in rows:
                row = row.split(",")
                badge_dict = {
                    "tag": row[0],
                    "url": row[1],
                    "slotName": row[2]
                }
                badges.append(to_class(ResoniteBadge, badge_dict))
        return badges

Class variables

var expire : datetime.datetime

var lastUpdate : datetime.datetime

var rememberMe : bool

var secretMachineIdHash : str

var secretMachineIdSalt : str

var session : requests.sessions.Session

var token : str

var userId : str

Static methods

def processRecordList(data: list[dict]) ‑> List[ResoniteRecord]

Processes a list of raw records and returns a list of ResoniteRecord objects.

Args

-----=

data

A list of dictionaries representing the raw records.

Returns -----= A list of ResoniteRecord objects.

Expand source code

@staticmethod
@deprecated_alias(process_record_ist)
def processRecordList(data: list[dict]) -> List[ResoniteRecord]:
    """ Processes a list of raw records and returns a list of ResoniteRecord objects.

    Args:
        data: A list of dictionaries representing the raw records.

    Returns:
        A list of ResoniteRecord objects.
    """
    return Client.process_record_ist(data=data)

def process_record_ist(data: List[dict]) ‑> List[ResoniteRecord]

Processes a list of raw records and returns a list of ResoniteRecord objects.

Args

-----=

data

A list of dictionaries representing the raw records.

Returns -----= A list of ResoniteRecord objects.

Expand source code

@staticmethod
def process_record_ist(data: List[dict]) -> List[ResoniteRecord]:
    """ Processes a list of raw records and returns a list of ResoniteRecord objects.

    Args:
        data: A list of dictionaries representing the raw records.

    Returns:
        A list of ResoniteRecord objects.
    """
    result = []
    for raw_item in data:
        item = to_class(ResoniteRecord, raw_item)
        record_class = recordTypeMapping.get(item.recordType, ResoniteRecord)
        x = to_class(record_class, raw_item)
        result.append(x)
    return result

Instance variables

var headers : dict

Returns the headers for API requests.

Returns

-----=

dict

A dictionary containing the headers.

Examples -----=

>>> client = Client()
>>> headers = client.headers

Expand source code

@property
def headers(self) -> dict:
    """ Returns the headers for API requests.

    Returns:
        dict: A dictionary containing the headers.

    Examples:
        >>> client = Client()
        >>> headers = client.headers
    """
    default = {"User-Agent": f"resonitepy/{__version__}"}
    if not self.userId or not self.token:
        logger.warning("Headers not properly set. API requests might throw an error soon...")
        return default
    default["Authorization"] = f"res {self.userId}:{self.token}"
    return default

Methods

def accept_contact_request(self, user_id: str)

Accept a pending contact request.

Expand source code

def accept_contact_request(self, user_id: str):
    """ Accept a pending contact request.
    """
    self._hub_operation(lambda hub: hub.accept_contact_request(user_id))

def add_contact(self, user_id: str)

Send a contact request to a user.

Expand source code

def add_contact(self, user_id: str):
    """ Send a contact request to a user.
    """
    self._hub_operation(lambda hub: hub.add_contact(user_id))

def badges(self) ‑> List[ResoniteBadge]

Return a list of ResoniteBadge object.

For now this is a simple list of hadcoded badges from a CSV file.

The badges with their tag surrounded [] are special badges which are not associated to a user directly but they are deduce from other data.

• [mobile]: user's platform is set to Mobile
• [linux]: user's platform is set to Linux
• [host]: user is the session host
• [<year>]: user's account was registred in ( would be something like 2020 or 2025 depending)
• [baguette]: user's account was registred at the same month and date as today but at a different year

Expand source code

def badges(self) -> List[ResoniteBadge]:
    """ Return a list of ResoniteBadge object.

    For now this is a simple list of hadcoded badges from a CSV file.

    The badges with their tag surrounded `[]` are special badges which are not
    associated to a user directly but they are deduce from other data.

    - `[mobile]`: user's platform is set to Mobile
    - `[linux]`: user's platform is set to Linux
    - `[host]`: user is the session host
    - `[<year>]`: user's account was registred in <year> (<year> would be something like 2020 or 2025 depending)
    - `[baguette]`: user's account was registred at the same month and date as today but at a different year

    """
    badges = []
    path = files("resonitepy").joinpath("data/badges.csv")
    with path.open(encoding="utf-8", newline="") as f:
        rows = (line for line in f if not line.lstrip().startswith("#"))
        for row in rows:
            row = row.split(",")
            badge_dict = {
                "tag": row[0],
                "url": row[1],
                "slotName": row[2]
            }
            badges.append(to_class(ResoniteBadge, badge_dict))
    return badges

def clean_session(self) ‑> None

Clean the client's session datas.

Returns -----= None

Examples -----=

>>> client = Client()
>>> client.clean_session()

Expand source code

def clean_session(self) -> None:
    """ Clean the client's session datas.

    Returns:
        None

    Examples:
        >>> client = Client()
        >>> client.clean_session()
    """
    self.userId = None
    self.token = None
    self.expire = None
    self.secretMachineIdHash = None
    self.secretMachineIdSalt = None
    self.lastUpdate = None

    if "Authorization" in self.session.headers:
        del self.session.headers["Authorization"]
    self.session.headers.update(self.headers)

def decline_contact_request(self, user_id: str)

Decline a pending contact request.

Expand source code

def decline_contact_request(self, user_id: str):
    """ Decline a pending contact request.
    """
    self._hub_operation(lambda hub: hub.decline_contact_request(user_id))

def getCloudVar(self, ownerId: str, path: str) ‑> ResoniteCloudVar

Retrieves a cloud variable for the specified owner and path.

Args

-----=

ownerId : str

The ID of the owner.

path : str

The path of the cloud variable.

Returns

-----=

ResoniteCloudVar

A ResoniteCloudVar object representing the cloud variable.

Examples -----=

>>> client = Client()
>>> cloud_var = client.getCloudVar('U-123', 'path/to/cloudvar')

Expand source code

def getCloudVar(self, ownerId: str, path: str) -> ResoniteCloudVar:
    """ Retrieves a cloud variable for the specified owner and path.

    Args:
        ownerId (str): The ID of the owner.
        path (str): The path of the cloud variable.

    Returns:
        ResoniteCloudVar: A ResoniteCloudVar object representing the cloud variable.

    Examples:
        >>> client = Client()
        >>> cloud_var = client.getCloudVar('U-123', 'path/to/cloudvar')
    """
    response = self.request(
        'get',
        f'/{self.getOwnerPath(ownerId)}/{ownerId}/vars/{path}'
    )
    return to_class(ResoniteCloudVar, response)

def getCloudVarDefs(self, ownerId: str, path: str) ‑> ResoniteCloudVarDefs

Retrieves the cloud variable definitions for the specified owner and path.

Args

-----=

ownerId : str

The ID of the owner.

path : str

The path of the cloud variable.

Returns

-----=

ResoniteCloudVarDefs

A ResoniteCloudVarDefs object representing the cloud variable definitions.

Raises

-----=

resonite_exceptions.ResoniteException

If the cloud variable doesn't exist.

Examples -----=

>>> client = Client()
>>> defs = client.getCloudVarDefs('U-123', 'path/to/cloudvar')

Expand source code

def getCloudVarDefs(self, ownerId: str, path: str) -> ResoniteCloudVarDefs:
    """ Retrieves the cloud variable definitions for the specified owner and path.

    Args:
        ownerId (str): The ID of the owner.
        path (str): The path of the cloud variable.

    Returns:
        ResoniteCloudVarDefs: A ResoniteCloudVarDefs object representing the cloud variable definitions.

    Raises:
        resonite_exceptions.ResoniteException: If the cloud variable doesn't exist.

    Examples:
        >>> client = Client()
        >>> defs = client.getCloudVarDefs('U-123', 'path/to/cloudvar')
    """
    json=[{
        "ownerId": ownerId,
        "path": path,
    }]

    response = self.request(
        'post',
        f'/readvars',
        json=json,
    )

    if not response:
        raise ResoniteException(f"{ownerId} {path} doesn't exist")

    return to_class(ResoniteCloudVarDefs, response[0]['definition'])

def getContacts(self) ‑> List[ResoniteContact]

Retrieves the contacts of the client.

Returns

-----=

List[ResoniteContact]

A list of ResoniteContact objects representing the contacts.

Examples -----=

>>> client = Client()
>>> contacts = client.getContacts()

Expand source code

@deprecated_alias(get_contacts)
def getContacts(self) -> List[ResoniteContact]:
    """ Retrieves the contacts of the client.

    Returns:
        List[ResoniteContact]: A list of ResoniteContact objects representing the contacts.

    Examples:
        >>> client = Client()
        >>> contacts = client.getContacts()
    """
    return self.get_contacts()

def getDirectory(self, directory: ResoniteDirectory) ‑> List[ResoniteRecord]

Retrieves the contents of a directory.

Args

-----=

directory : ResoniteDirectory

The ResoniteDirectory object representing the directory.

Returns

-----=

List[ResoniteRecord]

A list of ResoniteRecord objects representing the contents of the directory.

Examples -----=

>>> client = Client()
>>> directory = ResoniteDirectory(ownerId='U-123', content_path='path/to/directory')
>>> contents = client.getDirectory(directory)

Expand source code

def getDirectory(self, directory: ResoniteDirectory) -> List[ResoniteRecord]:
    """ Retrieves the contents of a directory.

    Args:
        directory (ResoniteDirectory): The ResoniteDirectory object representing the directory.

    Returns:
        List[ResoniteRecord]: A list of ResoniteRecord objects representing the contents of the directory.

    Examples:
        >>> client = Client()
        >>> directory = ResoniteDirectory(ownerId='U-123', content_path='path/to/directory')
        >>> contents = client.getDirectory(directory)
    """
    response = self.request(
        'get',
        f"/users/{directory.ownerId}/records",
        params={"path": directory.content_path},
    )
    return self.processRecordList(response)

def getGroup(self, groupId: str) ‑> ResoniteGroup

Retrieve group information.

Args

-----=

groupId : str

The group name starting with G-

Returns

-----=

ResoniteGroup

An object with the group information.

Expand source code

def getGroup(self, groupId: str) -> ResoniteGroup:
    """ Retrieve group information.

    Args:
        groupId (str): The group name starting with G-

    Returns:
        ResoniteGroup: An object with the group information.
    """
    response = self.request('get', f'/groups/{groupId}')
    return to_class(ResoniteGroup, response)

def getGroupMember(self, groupId: str, userId: str) ‑> ResoniteGroupMember

Retrieve a member from a group.

Args

-----=

groupId : str

The group name starting with G-

userId : str

The username starting with U-

Returns

-----=

ResoniteGroupMember

An object with the member information.

Expand source code

def getGroupMember(self, groupId: str, userId: str) -> ResoniteGroupMember:
    """ Retrieve a member from a group.

    Args:
        groupId (str): The group name starting with G-
        userId (str): The username starting with U-
    Returns:
        ResoniteGroupMember: An object with the member information.
    """
    response = self.request('get', f'/groups/{groupId}/members/{userId}')
    return to_class(ResoniteGroupMember, response)

def getGroupMembers(self, groupId: str) ‑> List[ResoniteGroupMember]

Retrieve members from a group.

Args

-----=

groupId : str

The group name starting with G-

Returns

-----=

List[ResoniteGroupMember]

A list with the group members.

Expand source code

def getGroupMembers(self, groupId: str) -> List[ResoniteGroupMember]:
    """ Retrieve members from a group.

    Args:
        groupId (str): The group name starting with G-

    Returns:
        List[ResoniteGroupMember]: A list with the group members.
    """
    response = self.request('get', f'/groups/{groupId}/members')
    return [to_class(ResoniteGroupMember, group_member) for group_member in response]

def getInventory(self) ‑> List[ResoniteRecord]

Retrieves the inventory of the user.

Returns

-----=

List[ResoniteRecord]

A list of ResoniteRecord objects representing the inventory.

Examples -----=

>>> client = Client()
>>> inventory = client.getInventory()

Expand source code

def getInventory(self) -> List[ResoniteRecord]:
    """ Retrieves the inventory of the user.

    Returns:
        List[ResoniteRecord]: A list of ResoniteRecord objects representing the inventory.

    Examples:
        >>> client = Client()
        >>> inventory = client.getInventory()
    """
    response = self.request(
        'get',
        f"/users/{self.userId}/records",
        params={"path": "Inventory"},
    )
    return self.processRecordList(response)

def getMemberships(self) ‑> List[ResoniteUserMembership]

Retrieve current connected user group memberships.

Return List[ResoniteUserMembership]: The list of groups where the user is a member.

Expand source code

def getMemberships(self) -> List[ResoniteUserMembership]:
    """ Retrieve current connected user group memberships.

    Return
        List[ResoniteUserMembership]: The list of groups where the user is a member.

    """
    if not self.userId:
        logger.error("Client not logged in")

    response = self.request('get', f'/users/{self.userId}/Memberships')
    return [to_class(ResoniteUserMembership, group) for group in response]

def getMessageLegacy(self, fromTime: str = None, maxItems: int = 100, user: str = None, unreadOnly: bool = False) ‑> List[ResoniteMessage]

Retrieves a list of Resonite messages.

This API endpoint should be understand as deprecated. Please use the SignalR protocol for this instead.

Args

-----=

fromTime : str

The starting time to retrieve messages from. (default: None, Not yet implemented)

maxItems : int

The maximum number of messages to retrieve. (default: 100)

user : str

The user ID to filter messages by. (default: None)

unreadOnly : bool

Whether to retrieve only unread messages. (default: False)

Returns -----= A list of ResoniteMessage objects.

Raises

-----=

ValueError

If fromTime is provided (not yet implemented).

Examples -----=

>>> client = ResoniteClient()
>>> messages = client.getMessageLegacy(maxItems=50, unreadOnly=True)

Expand source code

def getMessageLegacy(
    self,
    fromTime: str = None,
    maxItems: int = 100,
    user: str = None,
    unreadOnly: bool = False
) -> List[ResoniteMessage]:
    """ Retrieves a list of Resonite messages.

    This API endpoint should be understand as deprecated. Please use the SignalR
    protocol for this instead.

    Args:
        fromTime (str): The starting time to retrieve messages from. (default: None, Not yet implemented)
        maxItems (int): The maximum number of messages to retrieve. (default: 100)
        user (str): The user ID to filter messages by. (default: None)
        unreadOnly (bool): Whether to retrieve only unread messages. (default: False)

    Returns:
        A list of ResoniteMessage objects.

    Raises:
        ValueError: If fromTime is provided (not yet implemented).

    Examples:
        >>> client = ResoniteClient()
        >>> messages = client.getMessageLegacy(maxItems=50, unreadOnly=True)
    """
    if fromTime:
        raise ValueError('fromTime parameter is not yet implemented')

    if not self.userId:
        logger.error("Client not logged in")

    params = {
        "maxItems": maxItems,
        "unreadOnly": unreadOnly,
    }

    if user:
        params['user'] = user

    response = self.request(
        'get',
        f'/users/{self.userId}/messages',
        params=params
    )

    messages = []
    for message in response:
        messages.append(to_class(ResoniteMessage, message))

    return messages

def getOwnerPath(self, ownerId: str) ‑> str

Returns the owner path based on the owner ID.

Args

-----=

ownerId : str

The ID of the owner.

Returns

-----=

str

The owner path.

Raises

-----=

ValueError

If the owner type is invalid.

Examples -----=

>>> client = Client()
>>> owner_path = client.getOwnerPath('U-123')

Expand source code

def getOwnerPath(self, ownerId: str) -> str:
    """ Returns the owner path based on the owner ID.

    Args:
        ownerId (str): The ID of the owner.

    Returns:
        str: The owner path.

    Raises:
        ValueError: If the owner type is invalid.

    Examples:
        >>> client = Client()
        >>> owner_path = client.getOwnerPath('U-123')
    """
    ownerType = getOwnerType(ownerId)
    if ownerType == OwnerType.USER:
        return "users"
    elif ownerType == OwnerType.GROUP:
        return "groups"
    else:
        raise ValueError(f"invalid ownerType for {ownerId}")

def getSession(self, session_id: str) ‑> ResoniteSession

Retrieves session information for the specified session ID.

Args

-----=

session_id : str

The ID of the session.

Returns

-----=

ResoniteSession

A ResoniteSession object representing the session information.

Examples -----=

>>> client = Client()
>>> session = client.getSession('12345')

Expand source code

def getSession(self, session_id: str) -> ResoniteSession:
    """ Retrieves session information for the specified session ID.

    Args:
        session_id (str): The ID of the session.

    Returns:
        ResoniteSession: A ResoniteSession object representing the session information.

    Examples:
        >>> client = Client()
        >>> session = client.getSession('12345')
    """
    response = self.request('get', f'/sessions/{session_id}')
    return to_class(ResoniteSession, response)

def getSessions(self, compatibilityHash: str = None, name: str = None, universeId: str = None, hostName: str = None, hostId: str = None, minActiveUsers: int = 0, includeEmptyHeadless: bool = True) ‑> List[ResoniteSession]

Retrieves all active Resonite session.

Args

-----=

compatibilityHash : str

The compatibility-hash sessions needs to have. A Resonite client can only join a session if this hash matches between client and server.

name : str

The name of the session to search for.

universeId : str

The id of the universe sessions need to be part of.

hostName : str

The name of the user currently hosting the session.

hostId : str

The id of the user currently hosting the session.

minActiveUsers : int

he minimum amount of active users a session need to have. (default: 0)

includeEmptyHeadless

(bool): Should empty headless servers be included in the results. (Default: True)

Returns

-----=

List[ResoniteSession]

A list of ResoniteSessions

Examples -----=

>>> client = Client()
>>> client.getSession()

Expand source code

def getSessions(
    self,
    compatibilityHash: str = None,
    name: str = None,
    universeId: str = None,
    hostName: str = None,
    hostId: str = None,
    minActiveUsers: int = 0,
    includeEmptyHeadless: bool = True,
) -> List[ResoniteSession]:
    """ Retrieves all active Resonite session.

    Args:
        compatibilityHash (str): The compatibility-hash sessions needs to have. A Resonite client can only join a session if this hash matches between client and server.
        name (str): The name of the session to search for.
        universeId (str): The id of the universe sessions need to be part of.
        hostName (str): The name of the user currently hosting the session.
        hostId (str): The id of the user currently hosting the session.
        minActiveUsers (int): he minimum amount of active users a session need to have. (default: 0)
        includeEmptyHeadless: (bool): Should empty headless servers be included in the results. (Default: True)

    Returns:
        List[ResoniteSession]: A list of ResoniteSessions

    Examples:
        >>> client = Client()
        >>> client.getSession()
    """
    # TODO: Implement the search for the sessions
    response = self.request('get', '/sessions')
    return [to_class(ResoniteSession, session) for session in response]

def getUser(self, userId: str) ‑> ResoniteUser

Retrieve user directly.

Args

-----=

userId : int

Ther username starting with U-

Returns

-----=

ResoniteUser

The user

Expand source code

def getUser(self, userId: str) -> ResoniteUser:
    """ Retrieve user directly.

    Args:
        userId (int): Ther username starting with U-

    Returns:
        ResoniteUser: The user
    """
    response = self.request('get', f'/users/{userId}')
    return to_class(ResoniteUser, response)

def getUserByName(self, userId) ‑> ResoniteUser

Expand source code

def getUserByName(self, userId) -> ResoniteUser:
    reponse = self.request('get', f'/users/{userId}?byUsername=True')
    return to_class(ResoniteUser, reponse)

def getUserData(self, user: str = None) ‑> ResoniteUser

Retrieves user data for the specified user.

Args

-----=

user : str

The ID of the user to retrieve data for. If not provided, retrieves data for the client's user ID.

Returns

-----=

ResoniteUser

A ResoniteUser object representing the user data.

Examples -----=

>>> client = Client()
>>> user_data = client.getUserData()

Expand source code

def getUserData(self, user: str = None) -> ResoniteUser:
    """ Retrieves user data for the specified user.

    Args:
        user (str): The ID of the user to retrieve data for. If not provided, retrieves data for the client's user ID.

    Returns:
        ResoniteUser: A ResoniteUser object representing the user data.

    Examples:
        >>> client = Client()
        >>> user_data = client.getUserData()
    """
    if user is None:
        if not self.userId:
            logger.error("No user ID provided and client not logged in.")
        user = self.userId

    response = self.request('get', f"/users/{user}")
    return to_class(ResoniteUser, response)

def get_contacts(self, source: Literal['auto', 'hub', 'rest'] = 'auto', force: bool = False) ‑> List[ResoniteContact]

Retrieves the contacts of the client.

Returns

-----=

List[ResoniteContact]

A list of ResoniteContact objects representing the contacts.

Examples -----=

>>> client = Client()
>>> contacts = client.getContacts()

Expand source code

def get_contacts(self, source: Literal["auto", "hub", "rest"] = "auto", force: bool = False) -> List[ResoniteContact]:
    """ Retrieves the contacts of the client.

    Returns:
        List[ResoniteContact]: A list of ResoniteContact objects representing the contacts.

    Examples:
        >>> client = Client()
        >>> contacts = client.getContacts()
    """
    if source == "auto":
        source = "hub" if (self._hub is not None and self._hub.connected) else "rest"
    if source == "hub":
        return self._hub_operation(lambda hub: hub.get_contacts(force=force))
    response = self.request('get', f"/users/{self.userId}/contacts")
    return [to_class(ResoniteContact, user) for user in response]

def hub_close(self)

Disconnect the persistent hub and stop its thread.

Expand source code

def hub_close(self):
    """ Disconnect the persistent hub and stop its thread.
    """
    with self._hub_lock:
        if self._hub is not None and self._hub_loop is not None:
            try:
                asyncio.run_coroutine_threadsafe(
                    self._hub.disconnect(), self._hub_loop
                ).result(timeout=5)
            except Exception:
                pass
        if self._hub_loop is not None:
            self._hub_loop.call_soon_threadsafe(self._hub_loop.stop)
            self._hub_thread.join(timeout=5)
            self._hub_loop.close()
        self._hub = None
        self._hub_loop = None
        self._hub_thread = None
        atexit.unregister(self.hub_close)

def listCloudVar(self, ownerId: str) ‑> List[ResoniteCloudVar]

Lists the cloud variables for the specified owner.

Args

-----=

ownerId : str

The ID of the owner.

Returns

-----=

List[ResoniteCloudVar]

A list of ResoniteCloudVar objects representing the cloud variables.

Examples -----=

>>> client = Client()
>>> cloud_vars = client.listCloudVar('U-123')

Expand source code

def listCloudVar(self, ownerId: str) -> List[ResoniteCloudVar]:
    """ Lists the cloud variables for the specified owner.

    Args:
        ownerId (str): The ID of the owner.

    Returns:
        List[ResoniteCloudVar]: A list of ResoniteCloudVar objects representing the cloud variables.

    Examples:
        >>> client = Client()
        >>> cloud_vars = client.listCloudVar('U-123')
    """
    response = self.request(
        'get',
        f'/{self.getOwnerPath(ownerId)}/{ownerId}/vars'
    )
    return [to_class(ResoniteCloudVar, cloud_var) for cloud_var in response]

def load_token(self) ‑> None

Load the authentication token from a file.

Returns -----= None

Raises

-----=

ResoniteNoTokenError

If the token file does not exist or the token has expired.

Examples -----=

>>> client = Client()
>>> client.load_token()

Expand source code

def load_token(self) -> None:
    """ Load the authentication token from a file.

    Returns:
        None

    Raises:
        ResoniteNoTokenError: If the token file does not exist or the token has expired.

    Examples:
        >>> client = Client()
        >>> client.load_token()
    """
    if not path.exists(AUTHFILE_NAME):
        raise ResoniteNoTokenError("Auth token file not found.")

    with open(AUTHFILE_NAME, "r") as f:
        try:
            session = json.load(f)
            expire = datetime.fromisoformat(session.get("expire", ""))

            if datetime.now().timestamp() < expire.timestamp():
                self.token = session.get("token")
                self.userId = session.get("userId")
                self.expire = expire
                self.secretMachineIdHash = session.get("secretMachineIdHash")
                self.secretMachineIdSalt = session.get("secretMachineIdSalt")
                self.session.headers.update(self.headers)
            else:
                raise ResoniteNoTokenError
        except (json.JSONDecodeError, ValueError):
            raise ResoniteNoTokenError("Invalid auth token file format.")

def login(self, data: LoginDetails) ‑> None

Log in to the Resonite API with the provided login details.

Args

-----=

data : LoginDetails

The login details.

Returns -----= None

Examples -----=

>>> client = Client()
>>> login_details = LoginDetails(username='foxxie', password='pass')
>>> client.login(login_details)

Expand source code

def login(self, data: LoginDetails) -> None:
    """ Log in to the Resonite API with the provided login details.

    Args:
        data (LoginDetails): The login details.

    Returns:
        None

    Examples:
        >>> client = Client()
        >>> login_details = LoginDetails(username='foxxie', password='pass')
        >>> client.login(login_details)
    """
    payload = dataclasses.asdict(data)
    payload['authentication'] = data.authentication.build_dict()

    response = self.request('post', "/userSessions", json=payload)
    if not response:
        raise ResoniteException("Login failed - empty response")

    entity = response.get("entity", {})
    self.userId = entity.get("userId")
    self.token = entity.get("token")
    self.secretMachineIdHash = entity.get("secretMachineIdHash")
    self.secretMachineIdSalt = entity.get("secretMachineIdSalt")

    if "expire" in entity:
        self.expire = isoparse(entity["expire"])

    self.lastUpdate = datetime.now()
    self.session.headers.update(self.headers)

def logout(self) ‑> None

Log out the current session.

Returns -----= None

Examples -----=

>>> client = Client()
>>> client.logout()

Expand source code

def logout(self) -> None:
    """ Log out the current session.

    Returns:
        None

    Examples:
        >>> client = Client()
        >>> client.logout()
    """
    if self.userId and self.token:
        self.request(
            'delete',
            "/userSessions/{}/{}".format(self.userId, self.token),
            ignoreUpdate=True,
        )
    self.hub_close()
    self.clean_session()

def platform(self) ‑> Platform

Return information about the platform.

Expand source code

def platform(self) -> Platform:
    """ Return information about the platform.
    """
    response = self.request('get', '/platform')
    return to_class(Platform, response)

def remove_contact(self, user_id: str)

Remove a contact.

Expand source code

def remove_contact(self, user_id: str):
    """ Remove a contact.
    """
    self._hub_operation(lambda hub: hub.remove_contact(user_id))

def request(self, verb: str, path: str, data: dict = None, json: dict = None, params: dict = None, ignoreUpdate: bool = False) ‑> Dict

Sends an API request and returns the response.

While the API dont seems to implement more security, the official client behavior is respected. For now it will only disconnect after 1 day of inactivity.

Args

-----=

verb : str

The HTTP verb for the request.

path : str

The path of the API endpoint.

data : str

The data to send in the request body. (default: None)

json : dict

The JSON data to send in the request body. (default: None)

params : dict

The query parameters for the request. (default: None)

ignoreUpdate : bool

Whether to ignore the update check. (default: False)

Returns

-----=

Dict

The response from the API.

Raises

-----=

resonite_exceptions.InvalidCredentials

If the credentials are invalid.

resonite_exceptions.InvalidToken

If the token is invalid.

resonite_exceptions.ResoniteAPIException

If an API error occurs.

Example -----=

>>> client = Client()
>>> response = client.request('get', '/users/U-foxxie')

Expand source code

def request(
        self,
        verb: str,
        path: str,
        data: dict = None,
        json: dict = None,
        params: dict = None,
        ignoreUpdate: bool = False
    ) -> Dict:
    """ Sends an API request and returns the response.

    While the API dont seems to implement more security, the official client
    behavior is respected. For now it will only disconnect after 1 day of
    inactivity.

    Args:
        verb (str): The HTTP verb for the request.
        path (str): The path of the API endpoint.
        data (str): The data to send in the request body. (default: None)
        json (dict): The JSON data to send in the request body. (default: None)
        params (dict): The query parameters for the request. (default: None)
        ignoreUpdate (bool): Whether to ignore the update check. (default: False)

    Returns:
        Dict: The response from the API.

    Raises:
        resonite_exceptions.InvalidCredentials: If the credentials are invalid.
        resonite_exceptions.InvalidToken: If the token is invalid.
        resonite_exceptions.ResoniteAPIException: If an API error occurs.

    Example:
        >>> client = Client()
        >>> response = client.request('get', '/users/U-foxxie')
    """

    # Check if session needs to be refreshed
    if self.lastUpdate and not ignoreUpdate:
        lastUpdate = self.lastUpdate
        if (datetime.now() - lastUpdate).total_seconds() <= TOKEN_EXPIRY_SECONDS:
            self.request('patch', '/userSessions', ignoreUpdate=True)
            self.lastUpdate = datetime.now()
        # TODO: Implement disconnection after 1 week of inactivity when implementing the rememberMe feature.
        #if 64800 >= (datetime.now() - lastUpdate).total_seconds() >= 85536:
        #    self.request('patch', '/userSessions', ignoreUpdate=True)
        else:
            raise resonite_exceptions.InvalidToken("Token expired")

    # Prepare request arguments
    args = {'url': API_URL + path}
    if data: args['data'] = data
    if json: args['json'] = json
    if params: args['params'] = params

    # Execute the request
    func = getattr(self.session, verb, None)

    with func(**args) as req:
        logger.debug("ResoniteAPI: [{}] {}".format(req.status_code, args))

        # Handle error responses
        if req.status_code not in [200, 204]:
            if "Invalid credentials" in req.text:
                raise ResoniteInvalidCredentials(req.text)
            elif req.status_code == 403:
                raise ResoniteInvalidToken(req.headers)
            else:
                raise ResoniteAPIException(req)

        # Handle successful responses
        if req.status_code == 200:
            try:
                response = req.json()
                if "message" in response:
                    raise ResoniteAPIException(req, message=response["message"])
                return response
            except RequestsJSONDecodeError:
                return req.text

        # In case of a 204 response
        return

def resDBSignature(self, resUrl: str) ‑> str

Returns the Resonite DB signature from a Resonite URL.

Args

-----=

resUrl : url

The Resonite URL.

Returns

-----=

str

The Resonite DB signature.

Examples -----=

>>> client = Client()
>>> signature = client.resDBSignature("resrec://U-123/R-456")

Expand source code

@deprecated_alias(res_db_signature)
def resDBSignature(self, resUrl: str) -> str:
    """ Returns the Resonite DB signature from a Resonite URL.

    Args:
        resUrl (url): The Resonite URL.

    Returns:
        str: The Resonite DB signature.

    Examples:
        >>> client = Client()
        >>> signature = client.resDBSignature("resrec://U-123/R-456")
    """
    return self.res_db_signature(res_url=resUrl)

def resDbToHttp(self, resUrl: str) ‑> str

Converts a Resonite URL to an HTTP URL.

Args

-----=

resUrl : str

The Resonite URL.

Returns

-----=

str

The HTTP URL.

Examples -----=

>>> client = Client()
>>> http_url = client.resDbToHttp("resrec://U-123/R-456")

Expand source code

@deprecated_alias(res_db_to_http)
def resDbToHttp(self, resUrl: str) -> str:
    """  Converts a Resonite URL to an HTTP URL.

    Args:
        resUrl (str): The Resonite URL.

    Returns:
        str: The HTTP URL.

    Examples:
        >>> client = Client()
        >>> http_url = client.resDbToHttp("resrec://U-123/R-456")
    """
    return self.res_db_to_http(res_url=resUrl)

def res_db_signature(self, res_url: str) ‑> str

Returns the Resonite DB signature from a Resonite URL.

Args

-----=

reres_url : url

The Resonite URL.

Returns

-----=

str

The Resonite DB signature.

Examples -----=

>>> client = Client()
>>> signature = client.res_db_signature("resrec://U-123/R-456")

Expand source code

def res_db_signature(self, res_url: str) -> str:
    """ Returns the Resonite DB signature from a Resonite URL.

    Args:
        reres_url (url): The Resonite URL.

    Returns:
        str: The Resonite DB signature.

    Examples:
        >>> client = Client()
        >>> signature = client.res_db_signature("resrec://U-123/R-456")
    """
    parts = re.split("//+", res_url)
    if len(parts) > 2:
        raise ValueError(f"Invalid Resonite URL format: {res_url}")
    return parts[1].split(".")[0]

def res_db_to_http(self, res_url: str) ‑> str

Converts a Resonite URL to an HTTP URL.

Args

-----=

res_url : str

The Resonite URL.

Returns

-----=

str

The HTTP URL.

TODO: Fix example

Examples -----=

>>> client = Client()
>>> http_url = client.res_db_signature("resrec://U-123/R-456")

Expand source code

def res_db_to_http(self, res_url: str) -> str:
    """  Converts a Resonite URL to an HTTP URL.

    Args:
        res_url (str): The Resonite URL.

    Returns:
        str: The HTTP URL.

    # TODO: Fix example
    Examples:
        >>> client = Client()
        >>> http_url = client.res_db_signature("resrec://U-123/R-456")
    """
    return f"{ASSETS_URL.strip('/')}/{self.res_db_signature(res_url)}"

def resolveLink(self, link: ResoniteLink) ‑> ResoniteDirectory

Resolves a link type record and returns its directory.

Args

-----=

link : ResoniteLink

The ResoniteLink object representing the link type record.

Returns

-----=

ResoniteDirectory

A ResoniteDirectory object representing the directory.

Raises

-----=

resonite_exceptions.ResoniteException

If the link type is not supported.

resonite_exceptions.ResoniteAPIException

If the folder is not found in the cloud. Either delete or folder set back to non public. Supposed.

resonite_exceptions.InvalidToken

If denied permission to access the folder in the cloud. Supposed.

Examples -----=

>>> client = Client()
>>> link = ResoniteLink(assetUri=ParseResult(scheme='resrec', path='/G-Resonite/Inventory/Resonite Essentials'))  # This is not a valid ResoniteLink object but the minimal presend inside for this function to work
>>> directory = client.resolveLink(link)

Expand source code

def resolveLink(self, link: ResoniteLink) -> ResoniteDirectory:
    """ Resolves a link type record and returns its directory.

    Args:
        link (ResoniteLink): The ResoniteLink object representing the link type record.

    Returns:
        ResoniteDirectory: A ResoniteDirectory object representing the directory.

    Raises:
        resonite_exceptions.ResoniteException: If the link type is not supported.
        resonite_exceptions.ResoniteAPIException: If the folder is not found in the cloud. Either delete or folder set back to non public. Supposed.
        resonite_exceptions.InvalidToken: If denied permission to access the folder in the cloud. Supposed.

    Examples:
        >>> client = Client()
        >>> link = ResoniteLink(assetUri=ParseResult(scheme='resrec', path='/G-Resonite/Inventory/Resonite Essentials'))  # This is not a valid ResoniteLink object but the minimal presend inside for this function to work
        >>> directory = client.resolveLink(link)
    """
    if link.assetUri.scheme != 'resrec':
        raise resonite_exceptions.ResoniteException(f"Not supported scheme '{link.assetUri.scheme}' for link type {link}")

    owner_id = None
    record = None
    record_type = None
    record_path = None

    match_user_link_legacy = re.search(r'\/(U-.*)\/(R-.*)', link.assetUri.path)
    if not record_path and match_user_link_legacy:
        owner_id = match_user_link_legacy.group(1)
        record = match_user_link_legacy.group(2)
        record_type = "users"
        record_path = (
            link.assetUri.path
            .replace('/'+owner_id+'/', '')
            .replace('/', '\\')
        )

    match_group_link_legacy = re.search(r'\/(G-.*)\/(R-.*)', link.assetUri.path)
    if not record_path and match_group_link_legacy:
        owner_id = match_group_link_legacy.group(1)
        record = match_group_link_legacy.group(2)
        record_type = "groups"
        record_path = (
            link.assetUri.path
            .replace('/'+owner_id+'/', '')
            .replace('/', '\\')
        )

    if link.id and link.assetUri.path:
        match_user_link = re.search(r'\/(U-.*?)\/', link.assetUri.path)
        if not record_path and match_user_link:
            owner_id = match_user_link.group(1)
            record_type = "users"
            record_path = (
                link.assetUri.path
                .replace('/'+owner_id+'/', '')
                .replace('/', '\\')
            )
        match_group_link = re.search(r'\/(G-.*?)\/', link.assetUri.path)
        if not record_path and match_group_link:
            owner_id = match_group_link.group(1)
            record_type = "groups"
            record_path = (
                link.assetUri.path
                .replace('/'+owner_id+'/', '')
                .replace('/', '\\')
            )
        record = link.id

    if not owner_id or not record or not record_type:
        raise resonite_exceptions.ResoniteException(f'Not supported group type in link type {link}')

    response = self.request(
        'get',
        f"/{record_type}/{owner_id}/records/{record_path}",
    )
    return to_class(ResoniteDirectory, response)

def save_token(self) ‑> None

Saves the authentication token to a file.

Returns -----= None

Examples -----=

>>> client = Client()
>>> client.save_token()

Expand source code

def save_token(self) -> None:
    """ Saves the authentication token to a file.

    Returns:
        None

    Examples:
        >>> client = Client()
        >>> client.save_token()
    """
    if not all([self.userId, self.token, self.expire]):
        logger.warning("Cannot save token - missing required authentication data")
        return

    with open(AUTHFILE_NAME, "w+") as f:
        json.dump(
            {
                "userId": self.userId,
                "expire": self.expire.isoformat(),
                "token": self.token,
                "secretMachineIdHash": self.secretMachineIdHash,
                "secretMachineIdSalt": self.secretMachineIdSalt,
            },
            f,
        )

def searchUser(self, username: str) ‑> List[ResoniteUser]

Searches for users based on username.

This is not the U- Resonite user id, the API will search over usernames not ids.

Args

-----=

username : str

The username to search for.

Returns

-----=

List[ResoniteUser]

A list of ResoniteUser objects matching the search criteria.

Examples -----=

>>> client = Client()
>>> users = client.searchUser('foxxie')

Expand source code

def searchUser(self, username: str) -> List[ResoniteUser]:
    """ Searches for users based on username.

    This is not the U- Resonite user id, the API will search over usernames not ids.

    Args:
        username (str): The username to search for.

    Returns:
        List[ResoniteUser]: A list of ResoniteUser objects matching the search criteria.

    Examples:
        >>> client = Client()
        >>> users = client.searchUser('foxxie')
    """
    #TODO: the entitlements part could be optimized!
    response = self.request(
        'get',
        '/users',
        params = {'name': username}
    )
    users = []
    for user in response:
        users.append(to_class(ResoniteUser, user))
    return users

def setCloudVar(self, ownerId: str, path: str, value: str) ‑> None

Sets the value of a cloud variable for the specified owner and path.

Args

-----=

ownerId : str

The ID of the owner.

path : str

The path of the cloud variable.

value : str

The value to set for the cloud variable.

Returns -----= None

Examples -----=

>>> client = Client()
>>> client.setCloudVar('U-123', 'path/to/cloudvar', 'new value')

Expand source code

def setCloudVar(self, ownerId: str, path: str, value: str) -> None:
    """ Sets the value of a cloud variable for the specified owner and path.

    Args:
        ownerId (str): The ID of the owner.
        path (str): The path of the cloud variable.
        value (str): The value to set for the cloud variable.

    Returns:
        None

    Examples:
        >>> client = Client()
        >>> client.setCloudVar('U-123', 'path/to/cloudvar', 'new value')
    """
    return self.request(
        'put',
        f'/{self.getOwnerPath(ownerId)}/{ownerId}/vars/{path}',
        json = {
            "ownerId": ownerId,
            "path": path,
            "value": value,
        }
    )

def set_contact_status(self, user_id: str, status: ContactStatus, timeout: float = 5.0)

Set user side of the relationship to an explicit status. See HubManager.set_contact_status for what each status does.

Expand source code

def set_contact_status(self, user_id: str, status: ContactStatus, timeout: float = 5.0):
    """ Set user side of the relationship to an explicit status.
    See HubManager.set_contact_status for what each status does.
    """
    self._hub_operation(lambda hub: hub.set_contact_status(user_id, status, timeout))