feat(models): converted to pydantic for all data models
This commit is contained in:
parent
0dda9b46fa
commit
c6b723a476
GPG key ID:
B8953EC01E5C4063
6 changed files with 329 additions and 315 deletions
|
|
@ -1,64 +1,67 @@
|
|||
"""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
|
||||
from datetime import datetime
|
||||
from pyzipline.utils import convert_str_to_datetime
|
||||
from typing import Dict, List, Optional, Union
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class File:
|
||||
class File(BaseModel):
|
||||
"""File object used for uploading files to Zipline
|
||||
|
||||
Attributes:
|
||||
created_at (datetime.datetime): Datetime object of when the file was created
|
||||
createdAt (datetime.datetime): Datetime object of when the file was created
|
||||
id (int): ID of the file
|
||||
mimetype (str): String of the file's mimetype
|
||||
views (int): Integer of the number of views the file has
|
||||
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
|
||||
original_name (Optional[str]): String of the file's original name
|
||||
originalName (Optional[str]): String of the file's original name
|
||||
url (Optional[str]): String of the file's URL
|
||||
max_views (Optional[int]): Integer of the file's maximum number of views
|
||||
expired_at (Optional[datetime]): Datetime object of when the file will expire
|
||||
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
|
||||
folder_id (Optional[int]): Integer of the file's folder ID
|
||||
folderId (Optional[int]): Integer of the file's folder ID
|
||||
"""
|
||||
def __init__(
|
||||
self,
|
||||
createdAt: datetime,
|
||||
id: int, # pylint: disable=redefined-builtin
|
||||
mimetype: str,
|
||||
views: int,
|
||||
name: str,
|
||||
size: int,
|
||||
favorite: bool,
|
||||
originalName: str = None,
|
||||
url: str = None,
|
||||
maxViews: int = None,
|
||||
expiredAt: datetime = None,
|
||||
thumbnail: str = None,
|
||||
folderId: int = None,
|
||||
**kwargs
|
||||
):
|
||||
self.created_at = createdAt
|
||||
self.id = id
|
||||
self.mimetype = mimetype
|
||||
self.views = views
|
||||
self.name = name
|
||||
self.size = size
|
||||
self.favorite = favorite
|
||||
self.original_name = originalName
|
||||
self.url = url
|
||||
self.max_views = maxViews
|
||||
self.expired_at = expiredAt
|
||||
self.thumbnail = thumbnail
|
||||
self.folder_id = folderId
|
||||
self.__dict__.update(kwargs)
|
||||
createdAt: datetime
|
||||
id: int
|
||||
mimetype: str
|
||||
views: int
|
||||
name: str
|
||||
size: int
|
||||
favorite: bool
|
||||
originalName: Optional[str] = None
|
||||
url: Optional[str] = None
|
||||
maxViews: Optional[int] = None
|
||||
expiredAt: Optional[datetime] = None
|
||||
thumbnail: Optional[str] = None
|
||||
folderId: Optional[int] = None
|
||||
|
||||
def __str__(self):
|
||||
return self.name
|
||||
|
||||
class Invite(BaseModel):
|
||||
"""Invite object used for managing invites
|
||||
|
||||
class Result:
|
||||
Attributes:
|
||||
id (int): Integer ID of the invite
|
||||
code (str): String of the invite's code
|
||||
createdAt (datetime): Datetime object of when the invite was created
|
||||
expiresAt (datetime): Datetime object of when the invite will expire
|
||||
used (bool): Boolean of whether the invite has been used
|
||||
createdById (int): Integer ID of the user who created the invite
|
||||
"""
|
||||
id: int
|
||||
code: str
|
||||
createdAt: datetime
|
||||
expiresAt: datetime
|
||||
used: bool
|
||||
createdById: int
|
||||
|
||||
def __str__(self):
|
||||
return self.code
|
||||
|
||||
class Result(BaseModel):
|
||||
"""Result returned from low-level RestAdapter
|
||||
|
||||
Attributes:
|
||||
|
|
@ -67,50 +70,44 @@ class Result:
|
|||
message (str = ''): Human readable result
|
||||
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 {}
|
||||
success: bool
|
||||
status_code: int
|
||||
message: str = ''
|
||||
data: Union[List[Dict], Dict] = {}
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.status_code}: {self.message}\n{self.data}"
|
||||
return f"{self.status_code}: {self.message}"
|
||||
|
||||
|
||||
class Invite:
|
||||
"""Invite object used for managing invites
|
||||
class TypesCountItem(BaseModel):
|
||||
"""Model used in [StatsData](.#pyzipline.models.StatsData) for storing the number of files with a specific mimetype
|
||||
|
||||
Attributes:
|
||||
id (int): Integer ID of the invite
|
||||
code (str): String of the invite's code
|
||||
created_at (datetime): Datetime object of when the invite was created
|
||||
expires_at (datetime): Datetime object of when the invite will expire
|
||||
used (bool): Boolean of whether the invite has been used
|
||||
created_by_id (int): Integer ID of the user who created the invite
|
||||
mimetype (str): String of the mimetype
|
||||
count (int): Integer of the number of files with this mimetype
|
||||
"""
|
||||
def __init__(
|
||||
self,
|
||||
id: int, # pylint: disable=redefined-builtin
|
||||
code: str,
|
||||
createdAt: str,
|
||||
expiresAt: str,
|
||||
used: bool,
|
||||
createdById: int,
|
||||
**kwargs
|
||||
):
|
||||
self.id = id
|
||||
self.code = code
|
||||
self.created_at = convert_str_to_datetime(createdAt)
|
||||
self.expires_at = convert_str_to_datetime(expiresAt)
|
||||
self.used = used
|
||||
self.created_by_id = createdById
|
||||
self.__dict__.update(kwargs)
|
||||
count: int
|
||||
mimetype: str
|
||||
|
||||
def __str__(self):
|
||||
return self.code
|
||||
return f"{self.mimetype}: {self.count}"
|
||||
|
||||
class Stats:
|
||||
"""Stats object used for retrieving stats
|
||||
|
||||
class CountByUserItem(BaseModel):
|
||||
"""Model used in [StatsData](.#pyzipline.models.StatsData) 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
|
||||
"""
|
||||
count: int
|
||||
username: str
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.username}: {self.count}"
|
||||
|
||||
|
||||
class StatsData(BaseModel):
|
||||
"""Stats data model
|
||||
|
||||
Attributes:
|
||||
id (int): Integer ID of the stats
|
||||
|
|
@ -124,225 +121,64 @@ class Stats:
|
|||
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
|
||||
size: str
|
||||
count: int
|
||||
size_num: int
|
||||
count_users: int
|
||||
types_count: List[TypesCountItem]
|
||||
views_count: int
|
||||
count_by_user: List[CountByUserItem]
|
||||
|
||||
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
|
||||
class Stats(BaseModel):
|
||||
"""Stats model
|
||||
|
||||
Attributes:
|
||||
id (int): Integer ID of the OAuth
|
||||
provider (str): String of the OAuth's provider, one of 'DISCORD', 'GITHUB', 'GOOGLE'
|
||||
user_id (int): Integer ID of the user who owns the OAuth
|
||||
oauth_id (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
|
||||
id (int): Integer ID of the stats
|
||||
createdAt (datetime): Datetime object of when the stats were created
|
||||
data (StatsData): StatsData object of the stats data
|
||||
max_timestamp (Optional[datetime]): Datetime object of the maximum timestamp of the stats
|
||||
"""
|
||||
def __init__(
|
||||
self,
|
||||
id: int, # pylint: disable=redefined-builtin
|
||||
provider: str,
|
||||
oauthId: int,
|
||||
providerId: str,
|
||||
username: str,
|
||||
token: str,
|
||||
refresh: Optional[str],
|
||||
**kwargs
|
||||
):
|
||||
self.id = id
|
||||
self.provider = provider
|
||||
self.oauth_id = oauthId
|
||||
self.provider_id = providerId
|
||||
self.username = username
|
||||
self.token = token
|
||||
self.refresh = refresh
|
||||
self.__dict__.update(kwargs)
|
||||
|
||||
def __str__(self):
|
||||
return self.provider
|
||||
id: int
|
||||
createdAt: datetime
|
||||
data: StatsData
|
||||
max_timestamp: str = Field(default=None)
|
||||
|
||||
|
||||
class User:
|
||||
"""Object containing user information
|
||||
class Embed(BaseModel):
|
||||
color: Optional[str] = None
|
||||
title: Optional[str] = None
|
||||
siteName: Optional[str] = None
|
||||
description: Optional[str] = None
|
||||
|
||||
/// admonition | Contains Sensitive Information
|
||||
type: danger
|
||||
Please be mindful of how you use/store this object, as it contains sensitive information such as the user's token and OAuth/TOTP information.
|
||||
///
|
||||
|
||||
Attributes:
|
||||
id (int): Integer ID of the user
|
||||
uuid (str): String of the user's UUID
|
||||
username (str): String of the user's username
|
||||
avatar (Optional[str]): String of the user's avatar, base64 encoded
|
||||
token (str): String of the user's token
|
||||
administrator (bool): Boolean of whether the user is an administrator
|
||||
super_admin (bool): Boolean of whether the user is a super administrator
|
||||
system_theme (str): String of the user's system theme
|
||||
embed (Embed): Embed object of the user's embed
|
||||
totp_secret (Optional[str]): String of the user's TOTP secret
|
||||
domains (List[str]): List of Strings of the user's domains
|
||||
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,
|
||||
id: int, # pylint: disable=redefined-builtin
|
||||
uuid: str,
|
||||
username: str,
|
||||
avatar: Optional[str],
|
||||
token: str,
|
||||
administrator: bool,
|
||||
superAdmin: bool,
|
||||
systemTheme: str,
|
||||
embed: 'Embed',
|
||||
totpSecret: Optional[str],
|
||||
domains: List[str],
|
||||
oauth: Optional[List['OAuth']] = None,
|
||||
ratelimit: Optional[datetime] = None,
|
||||
**kwargs
|
||||
):
|
||||
self.id = id
|
||||
self.uuid = uuid
|
||||
self.username = username
|
||||
self.avatar = avatar
|
||||
self.token = token
|
||||
self.administrator = administrator
|
||||
self.super_admin = superAdmin
|
||||
self.system_theme = systemTheme
|
||||
self.embed = self.Embed(**embed)
|
||||
self.totp_secret = totpSecret
|
||||
self.domains = domains
|
||||
self.oauth = oauth
|
||||
self.ratelimit = ratelimit
|
||||
self.__dict__.update(kwargs)
|
||||
if self.oauth is not None:
|
||||
for oauth_entry in self.oauth:
|
||||
self.oauth.remove(oauth_entry)
|
||||
o = OAuth(**oauth_entry)
|
||||
self.oauth.append(o)
|
||||
class User(BaseModel):
|
||||
id: int
|
||||
uuid: str
|
||||
username: str
|
||||
avatar: Optional[str]
|
||||
token: str
|
||||
administrator: bool
|
||||
superAdmin: bool
|
||||
systemTheme: str
|
||||
embed: Embed
|
||||
ratelimit: None
|
||||
totpSecret: Optional[str]
|
||||
domains: List[str]
|
||||
|
||||
def __str__(self):
|
||||
return self.username
|
||||
|
||||
class Embed:
|
||||
"""Object containing a user's embed settings
|
||||
class Versions(BaseModel):
|
||||
stable: str
|
||||
upstream: str
|
||||
current: str
|
||||
|
||||
Attributes:
|
||||
color (Optional[str]): String of the embed's color
|
||||
title (Optional[str]): String of the embed's title
|
||||
site_name (Optional[str]): String of the embed's site name
|
||||
description (Optional[str]): String of the embed's description
|
||||
"""
|
||||
def __init__(
|
||||
self,
|
||||
color: str = None,
|
||||
title: str = None,
|
||||
siteName: str = None,
|
||||
description: str = None,
|
||||
**kwargs
|
||||
):
|
||||
self.color = color
|
||||
self.title = title
|
||||
self.site_name = siteName
|
||||
self.description = description
|
||||
self.__dict__.update(kwargs)
|
||||
|
||||
def __str__(self):
|
||||
if self.title is None:
|
||||
return "None"
|
||||
return self.title
|
||||
|
||||
class Version:
|
||||
"""Object containing the current, stable, and upstream versions of Zipline
|
||||
|
||||
Attributes:
|
||||
is_upstream (bool): Boolean of whether the current version is upstream (`trunk` branch)
|
||||
update_to_type (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.is_upstream = isUpstream
|
||||
self.update_to_type = updateToType
|
||||
self._versions = versions
|
||||
self.stable = self._versions['stable']
|
||||
self.upstream = self._versions['upstream']
|
||||
self.current = self._versions['upstream']
|
||||
class Version(BaseModel):
|
||||
isUpstream: bool
|
||||
updateToType: str
|
||||
versions: Versions
|
||||
|
||||
def __str__(self):
|
||||
return self.current
|
||||
return self.versions.current
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ from urllib3 import disable_warnings
|
|||
from pyzipline.exceptions import HTTPFailure, PyZiplineError
|
||||
from pyzipline.models import Result
|
||||
|
||||
|
||||
class RestAdapter:
|
||||
"""Constructor for RestAdapter
|
||||
|
||||
|
|
|
|||
|
|
@ -1,17 +1,19 @@
|
|||
"""This module contains the ZiplineApi class, which is the main class used to interact with the Zipline API."""
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Union, List
|
||||
import logging
|
||||
from datetime import datetime, timedelta
|
||||
from typing import List, Union
|
||||
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
|
||||
from pyzipline.exceptions import (FeatureDisabledError, Forbidden, NotFound,
|
||||
PyZiplineError)
|
||||
from pyzipline.models import File, Invite, Result, Stats, User, Version
|
||||
from pyzipline.rest_adapter import RestAdapter
|
||||
from pyzipline.exceptions import PyZiplineError, FeatureDisabledError, Forbidden, NotFound
|
||||
from pyzipline.models import User, File, Result, Invite, Stats, Version
|
||||
from pyzipline.utils import convert_datetime_to_str
|
||||
|
||||
# pylint: disable=not-a-mapping
|
||||
class ZiplineApi:
|
||||
"""Represents an instance of the Zipline API.
|
||||
|
||||
All API requests should be made through this class.
|
||||
class ZiplineApiConfig(BaseModel):
|
||||
"""Represents a configuration instance for the ZiplineApi class.
|
||||
|
||||
Args:
|
||||
hostname (str): The hostname of your Zipline instance, WITHOUT https or http.
|
||||
|
|
@ -20,15 +22,27 @@ class ZiplineApi:
|
|||
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.
|
||||
"""
|
||||
model_config = ConfigDict(arbitrary_types_allowed=True)
|
||||
hostname: str
|
||||
token: str = ''
|
||||
ssl: bool = True
|
||||
enforced_signing: bool = True
|
||||
logger: logging.Logger = logging.getLogger(__name__)
|
||||
|
||||
# pylint: disable=not-a-mapping
|
||||
class ZiplineApi:
|
||||
"""Represents an instance of the Zipline API.
|
||||
|
||||
All API requests should be made through this class.
|
||||
|
||||
Args:
|
||||
config (ZiplineApiConfig): Configuration object for the ZiplineApi class
|
||||
"""
|
||||
def __init__(
|
||||
self,
|
||||
hostname: str,
|
||||
token: str = '',
|
||||
ssl: bool = True,
|
||||
enforced_signing: bool = True,
|
||||
logger: logging.Logger = None
|
||||
config: ZiplineApiConfig
|
||||
):
|
||||
self._rest_adapter = RestAdapter(hostname=hostname, token=token, ssl=ssl, enforced_signing=enforced_signing, logger=logger)
|
||||
self._rest_adapter = RestAdapter(hostname=config.hostname, token=config.token, ssl=config.ssl, enforced_signing=config.enforced_signing, logger=config.logger)
|
||||
|
||||
def create_invite(self, expiry: timedelta = timedelta(days=1), count: int = 1) -> Union[Invite, List[Invite]]:
|
||||
"""Create an invite code
|
||||
|
|
@ -302,11 +316,7 @@ class ZiplineApi:
|
|||
"""
|
||||
result = self._rest_adapter.get(endpoint="auth/invite")
|
||||
if result.status_code == 200:
|
||||
invites = []
|
||||
for invite in result.data:
|
||||
i = Invite(**invite)
|
||||
invites.append(i)
|
||||
return invites
|
||||
return [Invite(**invite) for invite in result.data]
|
||||
if result.status_code == 401:
|
||||
raise Forbidden(result.message)
|
||||
if result.message == 'invites are disabled':
|
||||
|
|
@ -417,11 +427,7 @@ class ZiplineApi:
|
|||
"""
|
||||
result = self._rest_adapter.get(endpoint="users")
|
||||
if result.status_code == 200:
|
||||
users = []
|
||||
for user in result.data:
|
||||
u = User(**user)
|
||||
users.append(u)
|
||||
return users
|
||||
return [User(**user) for user in result.data]
|
||||
if result.status_code == 403:
|
||||
raise Forbidden(result.message)
|
||||
raise PyZiplineError(f"{result.status_code}: {result.message}\n{result.data}")
|
||||
|
|
@ -445,7 +451,7 @@ class ZiplineApi:
|
|||
///
|
||||
|
||||
Args:
|
||||
amount (int ): Number of stats to retrieve
|
||||
amount (int): Number of stats to retrieve
|
||||
force_update (bool): Force the Zipline instance to update its statistics before returning them, requires administrator
|
||||
|
||||
Raises:
|
||||
|
|
@ -463,14 +469,7 @@ class ZiplineApi:
|
|||
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)
|
||||
return [Stats(**stats) for stats in result.data]
|
||||
if result.status_code in (401, 403):
|
||||
raise Forbidden(result.message)
|
||||
raise PyZiplineError(f"{result.status_code}: {result.message}\n{result.data}")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue