feat: added get_stats and get_version, changed some stuff with models
Some checks failed
Pylint / Pylint (3.12) (push) Failing after 37s

This commit is contained in:
Seaswimmer 2023-12-21 14:08:55 -05:00
parent c5c28b4c4b
commit b6893d5077
Signed by: cswimr
GPG key ID: 1EBC234EEDA901AE
2 changed files with 240 additions and 42 deletions

View file

@ -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

View file

@ -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}")