PyZipline/pyzipline/zipline.py

229 lines
9.9 KiB
Python
Raw Normal View History

2023-12-20 23:51:31 -05:00
"""This module contains the ZiplineApi class, which is the main class used to interact with the Zipline API."""
from typing import Union, List
2023-12-19 05:36:18 -05:00
import logging
from pyzipline.rest_adapter import RestAdapter
2023-12-20 23:51:31 -05:00
from pyzipline.exceptions import PyZiplineError, FeatureDisabledError, Forbidden
from pyzipline.models import User, Result, Stats, Version
2023-12-19 05:36:18 -05:00
2023-12-20 23:51:31 -05:00
# pylint: disable=not-a-mapping
2023-12-19 05:36:18 -05:00
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.
2023-12-20 23:51:47 -05:00
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.
"""
2023-12-19 05:36:18 -05:00
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)
2023-12-20 23:51:57 -05:00
def register_user(self, username: str, password: str, invite: str = None, admin: bool = False) -> User:
"""Register a new user
2023-12-20 20:48:45 -05:00
2023-12-21 15:34:04 -05:00
/// 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:
2023-12-20 23:51:57 -05:00
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
2023-12-19 05:36:18 -05:00
2023-12-20 23:51:57 -05:00
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
2023-12-20 20:48:45 -05:00
Returns:
2023-12-20 23:51:57 -05:00
User: The newly created user
2023-12-19 05:36:18 -05:00
"""
2023-12-20 23:51:57 -05:00
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)
2023-12-20 20:48:45 -05:00
if result.status_code == 200:
return User(**result.data)
2023-12-20 23:51:57 -05:00
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
2023-12-21 15:34:04 -05:00
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
2023-12-20 20:48:45 -05:00
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
"""
2023-12-20 20:48:45 -05:00
data = {'username': username} if invite is None else {'username': username, 'code': invite}
2023-12-21 08:49:31 -05:00
result: Result = self._rest_adapter.post(endpoint="user/check", data=data)
if result.status_code == 200:
2023-12-20 20:48:45 -05:00
return False
2023-12-20 23:51:31 -05:00
if result.message == 'username already exists':
return True
2023-12-20 23:51:31 -05:00
if result.message == 'user registration is disabled':
2023-12-20 20:48:45 -05:00
raise FeatureDisabledError('user registration or invites are disabled')
2023-12-20 23:51:31 -05:00
if result.message == 'invalid invite code':
2023-12-20 20:48:45 -05:00
raise ValueError(result.message + "(most likely doesn't exist)")
2023-12-20 23:51:31 -05:00
if result.message == 'no code':
2023-12-20 20:48:45 -05:00
raise ValueError('invite code not provided')
2023-12-20 23:51:31 -05:00
if result.message == 'no username':
2023-12-20 20:48:45 -05:00
raise ValueError('username not provided')
2023-12-20 23:51:31 -05:00
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:
2023-12-21 14:08:09 -05:00
Forbidden: The user is not authenticated
2023-12-20 23:51:31 -05:00
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}")