PyZipline/pyzipline/zipline.py

"""This module contains the ZiplineApi class, which is the main class used to interact with the Zipline API."""
from typing import Union, List
import logging
from pyzipline.rest_adapter import RestAdapter
from pyzipline.exceptions import PyZiplineError, FeatureDisabledError, Forbidden
from pyzipline.models import User, Result, Stats, Version

# pylint: disable=not-a-mapping
class ZiplineApi:
    """Represents an instance of the Zipline API.

    All API requests should be made through this class.

    Args:
        hostname (str): The hostname of your Zipline instance, WITHOUT https or http.
        token (str): String used for authentication when making requests.
        ssl (bool): Normally set to True, but if your Zipline instance doesn't use SSL/TLS, set this to False.
        enforced_signing (bool): Normally set to True, but if having SSL/TLS cert validation issues, can turn off with False.
        logger (logging.Logger): If your app has a logger, pass it in here.
    """
    def __init__(
        self,
        hostname: str,
        token: str = '',
        ssl: bool = True,
        enforced_signing: bool = True,
        logger: logging.Logger = None
    ):
        self._rest_adapter = RestAdapter(hostname=hostname, token=token, ssl=ssl, enforced_signing=enforced_signing, logger=logger)

    def register_user(self, username: str, password: str, invite: str = None, admin: bool = False) -> User:
        """Register a new user

        /// admonition | Requires Authentication
            type: warning
        ///

        /// admonition | Conditionally Requires Administrator
            type: danger
        The authenticated user must be an Administrator to use the `admin` parameter or register a user when registration is disabled.
        ///

        Args:
            username (str): Username to register
            password (str): Password for the new user
            invite (str): Invite code to register the new user with, only required if registration without invites is disabled and the authenticated user is not an administrator
            admin (bool): Whether or not the new user should be an administrator, authenticated user must be a super administrator to create an administrator

        Raises:
            Forbidden: Raised if the authenticated user is not an super administrator and attempts to create an administrator
            FeatureDisabledError: Raised when:\n
              - registration or invites are disabled on the Zipline instance and the authenticated user is not an administrator
              - invite code is provided and invites are disabled
            PyZiplineError: Raised if the API changes, causing a breaking change in this method
            ValueError: Raised when the username is already taken or if the invite code is invalid/expired

        Returns:
            User: The newly created user
        """
        data = {'username': username, 'password': password}
        if invite is not None:
            data['code'] = invite
        if admin:
            data['admin'] = True

        result: Result = self._rest_adapter.post(endpoint="auth/register", data=data)
        if result.status_code == 200:
            return User(**result.data)

        if result.message == 'This endpoint is unavailable due to current configurations':
            raise FeatureDisabledError('user registration or invites are disabled')

        if result.message =='Bad Username/Password':
            if self.check_user_exists(username):
                raise ValueError('username already taken')
            raise FeatureDisabledError('invite code is provided and invites are disabled')

        if result.message == 'Bad invite':
            raise ValueError('invite code is invalid or expired')

        if result.message == 'not an administrator':
            raise Forbidden(result.message)

        raise PyZiplineError(f"{result.status_code}: {result.message}\n{result.data}")

    def check_user_exists(self, username: str, invite: str = None) -> bool:
        """Check if a user exists by username

        Args:
            username (str): Username to check
            invite (str): Invite code to use, only required if registration without invites is disabled

        Raises:
            FeatureDisabledError: Raised when registration or invites are disabled on the Zipline instance
            PyZiplineError: Raised if the API changes, causing a breaking change in this method
            ValueError: Raised when the username is not present, or the invite code is invalid/not present and invites are enabled

        Returns:
            bool: True if user exists, False if not
        """
        data = {'username': username} if invite is None else {'username': username, 'code': invite}
        result: Result = self._rest_adapter.post(endpoint="user/check", data=data)
        if result.status_code == 200:
            return False
        if result.message == 'username already exists':
            return True
        if result.message == 'user registration is disabled':
            raise FeatureDisabledError('user registration or invites are disabled')
        if result.message == 'invalid invite code':
            raise ValueError(result.message + "(most likely doesn't exist)")
        if result.message == 'no code':
            raise ValueError('invite code not provided')
        if result.message == 'no username':
            raise ValueError('username not provided')
        raise PyZiplineError(f"{result.status_code}: {result.message}\n{result.data}")

    def get_self(self) -> User:
        """Get the currently authenticated user

        /// admonition | Requires Authentication
            type: warning
        ///

        Raises:
            Forbidden: The user is not authenticated
            PyZiplineError: Raised if the API changes, causing a breaking change in this method

        Returns:
            User: The currently authenticated user
        """
        result = self._rest_adapter.get(endpoint="user")
        if result.status_code == 200:
            return User(**result.data)
        if result.status_code == 401:
            raise Forbidden(result.message)
        raise PyZiplineError(f"{result.status_code}: {result.message}\n{result.data}")

    def get_user(self, user_id: int) -> User:
        """Get a user by ID

        /// admonition | Requires Administrator
            type: danger
        ///

        Args:
            user_id (int): Integer ID of the user to retrieve

        Raises:
            Forbidden: Raised if the authenticated user is not an administrator
            PyZiplineError: Raised if the API changes, causing a breaking change in this method

        Returns:
            User: The user with the given ID
        """
        result = self._rest_adapter.get(endpoint=f"user/{user_id}")
        if result.status_code == 200:
            return User(**result.data)
        if result.status_code == 403:
            raise Forbidden(result.message)
        raise PyZiplineError(f"{result.status_code}: {result.message}\n{result.data}")

    def get_stats(self, amount: int = 1, force_update: bool = False) -> Union[Stats, List[Stats]]:
        """Get statistics about the Zipline instance

        /// admonition | Requires Authentication
            type: warning
        ///

        /// admonition | Parameter Requires Administrator
            type: danger
        The authenticated user must be an administrator to use the `force_update` argument.
        ///

        /// admonition | Configuration Varies
            type: note
        The endpoint this method uses, `/api/stats`, relies a lot on Zipline's [configuration](https://zipline.diced.sh/docs/config/website#website_show_files_per_user) to determine who can access the endpoint and what the endpoint returns depending on permission level.
        Please bear this in mind when using this method.
        ///

        Args:
            amount (int ): Number of stats to retrieve
            force_update (bool): Force the Zipline instance to update its statistics before returning them, requires administrator

        Raises:
            Forbidden: The user is not authenticated, or the user is not an administrator and `force_update` is True
            PyZiplineError: Raised if the API changes, causing a breaking change in this method
            ValueError: Raised if amount is less than 1

        Returns:
            Stats: Statistics about the Zipline instance"""
        if amount < 1:
            raise ValueError('amount must be greater than 0')
        if force_update:
            result = self._rest_adapter.post(endpoint="stats", params={'amount': amount})
        else:
            result = self._rest_adapter.get(endpoint="stats", params={'amount': amount})
        if result.status_code == 200:
            if amount > 1:
                stats_list = []
                for stats in result.data:
                    s = Stats(**stats)
                    stats_list.append(s)
                return stats_list
            data = result.data[0] if isinstance(result.data, list) else result.data
            return Stats(**data)
        if result.status_code == 401 or result.status_code == 403:
            raise Forbidden(result.message)
        raise PyZiplineError(f"{result.status_code}: {result.message}\n{result.data}")

    def get_version(self) -> Version:
        """Get the version of the Zipline instance

        /// admonition | Requires Authentication
            type: warning
        ///

        Raises:
            Forbidden: The user is not authenticated
            PyZiplineError: Raised if the API changes, causing a breaking change in this method

        Returns:
            Version: The version of the Zipline instance"""
        result = self._rest_adapter.get(endpoint="version")
        if result.status_code == 200:
            return Version(**result.data)
        if result.status_code == 401:
            raise Forbidden(result.message)
        raise PyZiplineError(f"{result.status_code}: {result.message}\n{result.data}")