feat: added check_user_exists() method
Some checks failed
Pylint / Pylint (3.12) (push) Failing after 37s
Some checks failed
Pylint / Pylint (3.12) (push) Failing after 37s
This commit is contained in:
parent
6a250a24fb
commit
5a8fe5451b
2 changed files with 33 additions and 19 deletions
|
@ -53,11 +53,7 @@ class RestAdapter:
|
|||
full_url = self._url + endpoint
|
||||
headers = {'Authorization': self._token}
|
||||
|
||||
log_line_pre = f"method={http_method}, url={full_url}, params={params}"
|
||||
log_line_post = ', '.join((log_line_pre, "success={}, status_code={}, message={}"))
|
||||
|
||||
try: # Log HTTP params and perform an HTTP request, catching and re-raising any exceptions
|
||||
self._logger.debug(msg=log_line_pre)
|
||||
try: # Perform an HTTP request, catching and re-raising any exceptions
|
||||
# will eventually refactor this to use asyncio/aiohttp instead for async operation
|
||||
response = requests.request(method=http_method, url=full_url, verify=self._enforced_signing, params=params, headers=headers, json=data)
|
||||
except requests.exceptions.RequestException as e:
|
||||
|
@ -67,14 +63,15 @@ class RestAdapter:
|
|||
try: # Deserialize JSON output to Python object, or return failed Result on exception
|
||||
data_out = response.json()
|
||||
except (ValueError, JSONDecodeError) as e:
|
||||
self._logger.error(msg=log_line_post.format(False, None, e))
|
||||
raise PyZiplineError("Could not decode response from Zipline server") from e
|
||||
|
||||
# If status_code in 200-299 range, return success Result with data, otherwise return failed Result with message
|
||||
is_success = 299 >= response.status_code >= 200
|
||||
|
||||
self._logger.debug(msg=log_line_post.format(is_success, response.status_code, response.reason))
|
||||
return Result(success=is_success, status_code=response.status_code, message=response.reason, data=data_out)
|
||||
if is_success:
|
||||
return Result(success=is_success, status_code=response.status_code, message=response.reason, data=data_out)
|
||||
else:
|
||||
return Result(success=is_success, status_code=response.status_code, message=data_out['error'], data=data_out)
|
||||
|
||||
def get(self, endpoint: str, params: Dict = None) -> Result:
|
||||
"""Make a GET request to the Zipline server. You should almost never have to use this directly.
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
import logging
|
||||
import json
|
||||
from pyzipline.rest_adapter import RestAdapter
|
||||
from pyzipline.errors import PyZiplineError, FeatureDisabledError
|
||||
from pyzipline.exceptions import PyZiplineError, FeatureDisabledError
|
||||
from pyzipline.models import *
|
||||
|
||||
class ZiplineApi:
|
||||
|
@ -28,11 +29,15 @@ class ZiplineApi:
|
|||
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
|
||||
|
||||
Returns:
|
||||
:class:`pyzipline.models.User`: The user with the given ID
|
||||
User: The user with the given ID
|
||||
"""
|
||||
result = self._rest_adapter.get(endpoint=f"user/{user_id}")
|
||||
return User(**result.data)
|
||||
|
@ -40,11 +45,18 @@ class ZiplineApi:
|
|||
def get_self(self) -> User:
|
||||
"""Get the currently authenticated user
|
||||
|
||||
/// admonition | Requires Authentication
|
||||
type: warning
|
||||
///
|
||||
|
||||
Returns:
|
||||
:class:`pyzipline.models.User`: The currently authenticated user
|
||||
User: The currently authenticated user
|
||||
"""
|
||||
result = self._rest_adapter.get(endpoint=f"user")
|
||||
return User(**result.data)
|
||||
if result.status_code == 200:
|
||||
return User(**result.data)
|
||||
elif result.status_code == 401:
|
||||
raise ValueError(result.message)
|
||||
|
||||
def check_user_exists(self, username: str, invite: str = None) -> bool:
|
||||
"""Check if a user exists by username
|
||||
|
@ -55,20 +67,25 @@ class ZiplineApi:
|
|||
|
||||
Raises:
|
||||
FeatureDisabledError: Raised when registration or invites are disabled on the Zipline instance
|
||||
PyZiplineError: Raised
|
||||
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
|
||||
"""
|
||||
params = {'username': username} if invite is None else {'username': username, 'code': invite}
|
||||
result: Result = self._rest_adapter.get(endpoint=f"user/check", params=params)
|
||||
data = {'username': username} if invite is None else {'username': username, 'code': invite}
|
||||
result: Result = self._rest_adapter.post(endpoint=f"user/check", data=data)
|
||||
if result.status_code == 200:
|
||||
return bool(result.data['success'])
|
||||
elif result.message == 'user resistration is disabled' or result.message == 'invites are disabled':
|
||||
raise FeatureDisabledError(result.message)
|
||||
return False
|
||||
elif result.message == 'username already exists':
|
||||
return True
|
||||
elif result.message == 'user registration is disabled':
|
||||
raise FeatureDisabledError('user registration or invites are disabled')
|
||||
elif result.message == 'invalid invite code':
|
||||
raise ValueError(result.message)
|
||||
raise ValueError(result.message + "(most likely doesn't exist)")
|
||||
elif result.message == 'no code':
|
||||
raise ValueError('invite code not provided')
|
||||
elif result.message == 'no username':
|
||||
raise ValueError('username not provided')
|
||||
else:
|
||||
raise PyZiplineError(result.message)
|
||||
|
|
Loading…
Reference in a new issue