From b6893d50772377d0af533e4b2cc2accaecb92131 Mon Sep 17 00:00:00 2001 From: SeaswimmerTheFsh Date: Thu, 21 Dec 2023 14:08:55 -0500 Subject: [PATCH] feat: added get_stats and get_version, changed some stuff with models --- pyzipline/models.py | 211 ++++++++++++++++++++++++++++++++++--------- pyzipline/zipline.py | 71 ++++++++++++++- 2 files changed, 240 insertions(+), 42 deletions(-) diff --git a/pyzipline/models.py b/pyzipline/models.py index a398a64..e4d81c6 100644 --- a/pyzipline/models.py +++ b/pyzipline/models.py @@ -1,32 +1,8 @@ """This is a list of all the models used in PyZipline. They are used to represent the data returned from the Zipline API.""" -from typing import List, Dict, Optional, Union +from typing import List, Dict, Optional from datetime import datetime -class Embed: - """Embed object used for checking embeds - - Attributes: - color (Optional[str]): String of the embed's color - title (Optional[str]): String of the embed's title - siteName (Optional[str]): String of the embed's site name - description (Optional[str]): String of the embed's description - """ - def __init__( - self, - color: str, - title: str, - siteName: str, - description: str, - **kwargs - ): - self.color = color - self.title = title - self.siteName = siteName - self.description = description - self.__dict__.update(kwargs) - - class File: """File object used for uploading files to Zipline @@ -38,12 +14,12 @@ class File: name (str): String of the file's name size (int): Integer of the file's size in bytes favorite (bool): Boolean of whether the file is favorited - originalName (str = None): (optional) String of the file's original name - url (str = None): (optional) String of the file's URL - maxViews (int = None): (optional) Integer of the file's maximum number of views - expiredAt (datetime.datetime = None): (optional) Datetime object of when the file will expire - thumbnail (str = None): (optional) String of the file's thumbnail URL - folderId (int = None): (optional) Integer of the file's folder ID + originalName (Optional[str]): String of the file's original name + url (Optional[str]): String of the file's URL + maxViews (Optional[int]): Integer of the file's maximum number of views + expiredAt (Optional[datetime]): Datetime object of when the file will expire + thumbnail (Optional[str]): String of the file's thumbnail URL + folderId (Optional[int]): Integer of the file's folder ID """ def __init__( self, @@ -57,7 +33,7 @@ class File: originalName: str = None, url: str = None, maxViews: int = None, - expiredAt: Union[datetime, None] = None, + expiredAt: datetime = None, thumbnail: str = None, folderId: int = None, **kwargs @@ -77,6 +53,9 @@ class File: self.folderId = folderId self.__dict__.update(kwargs) + def __str__(self): + return self.name + class Result: """Result returned from low-level RestAdapter @@ -85,13 +64,16 @@ class Result: success (bool): Boolean of whether the request was successful status_code (int): Standard HTTP Status code message (str = ''): Human readable result - data (List[Dict] = None): Python List of Dictionaries (or maybe just a single Dictionary on error) + data (Union[List[Dict], Dict]): Python List of Dictionaries (or maybe just a single Dictionary on error) """ def __init__(self, success: bool, status_code: int, message: str = '', data: List[Dict] = None): self.success = success self.status_code = status_code self.message = message - self.data = data if data else [] + self.data = data if data else {} + + def __str__(self): + return f"{self.status_code}: {self.message}\n{self.data}" class Invite: @@ -123,6 +105,93 @@ class Invite: self.createdById = createdById self.__dict__.update(kwargs) + def __str__(self): + return self.code + +class Stats: + """Stats object used for retrieving stats + + Attributes: + id (int): Integer ID of the stats + createdAt (datetime): Datetime object of when the stats were created + max_timestamp (Optional[datetime]): Datetime object of the maximum timestamp of the stats + size (str): String of the size of the files + size_num (int): Integer of the size of the files in bytes + count (int): Integer of the number of files + count_users (int): Integer of the number of users + views_count (int): Integer of the number of views + types_count (Optional[List[Mimetype]]): List of Mimetype objects + count_by_user (Optional[List[CountByUser]]): List of CountByUser objects + """ + def __init__( + self, + id: int, # pylint: disable=redefined-builtin + createdAt: datetime, + data: dict, + max_timestamp: Optional[datetime] = None + ): + self.id = id + self.createdAt = createdAt + self._data = data + self.max_timestamp = max_timestamp + self.size = self._data['size'] + self.size_num = self._data['size_num'] + self.count = self._data['count'] + self.count_users = self._data['count_users'] + self.views_count = self._data['views_count'] + self.types_count: list = self._data['types_count'] + self.count_by_user: list = self._data['count_by_user'] + if self.types_count is not None: + new_types_count = [] + for mimetype_entry in self.types_count: + if isinstance(mimetype_entry, dict): + m = self.Mimetype(**mimetype_entry) + new_types_count.append(m) + self.types_count = new_types_count + if self.count_by_user is not None: + new_count_by_user = [] + for count_by_user_entry in self.count_by_user: + if isinstance(count_by_user_entry, dict): + c = self.CountByUser(**count_by_user_entry) + new_count_by_user.append(c) + self.count_by_user = new_count_by_user + + def __str__(self): + return str(self.id) + + class Mimetype: + """Object used in [Stats](.#pyzipline.models.Stats) for storing the number of files with a specific mimetype + + Attributes: + mimetype (str): String of the mimetype + count (int): Integer of the number of files with this mimetype""" + def __init__( + self, + mimetype: str, + count: int + ): + self.mimetype = mimetype + self.count = count + + def __str__(self): + return f"{self.mimetype}: {self.count}" + + class CountByUser: + """Object used in [Stats](.#pyzipline.models.Stats) for storing the number of files uploaded by a user + + Attributes: + username (str): String of the username + count (int): Integer of the number of files uploaded by this user""" + def __init__( + self, + username: str, + count: int + ): + self.username = username + self.count = count + + def __str__(self): + return f"{self.username}: {self.count}" class OAuth: """OAuth object used for managing OAuth @@ -131,7 +200,7 @@ class OAuth: id (int): Integer ID of the OAuth provider (str): String of the OAuth's provider, one of 'DISCORD', 'GITHUB', 'GOOGLE' userId (int): Integer ID of the user who owns the OAuth - providerId (str): String of the OAuth's provider ID + oauthId (str): String of the OAuth's provider ID username (str): String of the OAuth's connected account's username token (str): String of the OAuth's access token refresh (Optional[str]): String of the OAuth's refresh token @@ -140,7 +209,7 @@ class OAuth: self, id: int, # pylint: disable=redefined-builtin provider: str, - userId: int, + oauthId: int, providerId: str, username: str, token: str, @@ -149,13 +218,16 @@ class OAuth: ): self.id = id self.provider = provider - self.userId = userId + self.oauthId = oauthId self.providerId = providerId self.username = username self.token = token self.refresh = refresh self.__dict__.update(kwargs) + def __str__(self): + return self.provider + class User: """User object used for managing users @@ -172,8 +244,8 @@ class User: embed (Embed): Embed object of the user's embed totpSecret (Optional[str]): String of the user's TOTP secret domains (List[str]): List of Strings of the user's domains - oauth (Optional[List[OAuth]] = None): List of [OAuth](.#pyzipline.models.OAuth) objects - ratelimit (Optional[datetime] = None): Datetime object of when the user's ratelimit expires + oauth (Optional[List[OAuth]]): List of [OAuth](.#pyzipline.models.OAuth) objects + ratelimit (Optional[datetime]): Datetime object of when the user's ratelimit expires """ def __init__( self, @@ -185,7 +257,7 @@ class User: administrator: bool, superAdmin: bool, systemTheme: str, - embed: Embed, + embed: 'Embed', totpSecret: Optional[str], domains: List[str], oauth: Optional[List['OAuth']] = None, @@ -200,7 +272,7 @@ class User: self.administrator = administrator self.superAdmin = superAdmin self.systemTheme = systemTheme - self.embed = Embed(**embed) + self.embed = self.Embed(**embed) self.totpSecret = totpSecret self.domains = domains self.oauth = oauth @@ -211,3 +283,60 @@ class User: self.oauth.remove(oauth_entry) o = OAuth(**oauth_entry) self.oauth.append(o) + + def __str__(self): + return self.username + + class Embed: + """Embed object used for checking embeds + + Attributes: + color (Optional[str]): String of the embed's color + title (Optional[str]): String of the embed's title + siteName (Optional[str]): String of the embed's site name + description (Optional[str]): String of the embed's description + """ + def __init__( + self, + color: str, + title: str, + siteName: str, + description: str, + **kwargs + ): + self.color = color + self.title = title + self.siteName = siteName + self.description = description + self.__dict__.update(kwargs) + + def __str__(self): + if self.title is None: + return "None" + return self.title + +class Version: + """Version object containing the current, stable, and upstream versions of Zipline + + Attributes: + isUpstream (bool): Boolean of whether the current version is upstream + updateToType (str): String of the type of update available, one of 'stable' or 'upstream' + stable (str): String of the stable version + upstream (str): String of the upstream version + current (str): String of the current version + """ + def __init__( + self, + isUpstream: bool, + updateToType: str, + versions: {dict} + ): + self.isUpstream = isUpstream + self.updateToType = updateToType + self._versions = versions + self.stable = self._versions['stable'] + self.upstream = self._versions['upstream'] + self.current = self._versions['upstream'] + + def __str__(self): + return self.current diff --git a/pyzipline/zipline.py b/pyzipline/zipline.py index 7ba38b3..12ae460 100644 --- a/pyzipline/zipline.py +++ b/pyzipline/zipline.py @@ -1,8 +1,9 @@ """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 * # pylint: disable=wildcard-import,unused-wildcard-import +from pyzipline.models import User, Result, Stats, Version # pylint: disable=not-a-mapping class ZiplineApi: @@ -148,3 +149,71 @@ class ZiplineApi: 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}")