feat: added get_password_protected_file method
Some checks failed
Pylint / Pylint (3.12) (push) Failing after 36s

This commit is contained in:
Seaswimmer 2023-12-22 21:01:59 -05:00
parent 0de2fcd024
commit ec9509e5ea
Signed by: cswimr
GPG key ID: 1EBC234EEDA901AE
4 changed files with 51 additions and 5 deletions

20
poetry.lock generated
View file

@ -878,6 +878,24 @@ files = [
{file = "mkdocs_material_extensions-1.3.1.tar.gz", hash = "sha256:10c9511cea88f568257f960358a467d12b970e1f7b2c0e5fb2bb48cab1928443"},
]
[[package]]
name = "mkdocs-redirects"
version = "1.2.1"
description = "A MkDocs plugin for dynamic page redirects to prevent broken links."
optional = false
python-versions = ">=3.6"
files = [
{file = "mkdocs-redirects-1.2.1.tar.gz", hash = "sha256:9420066d70e2a6bb357adf86e67023dcdca1857f97f07c7fe450f8f1fb42f861"},
]
[package.dependencies]
mkdocs = ">=1.1.1"
[package.extras]
dev = ["autoflake", "black", "isort", "pytest", "twine (>=1.13.0)"]
release = ["twine (>=1.13.0)"]
test = ["autoflake", "black", "isort", "pytest"]
[[package]]
name = "mkdocstrings"
version = "0.24.0"
@ -1663,4 +1681,4 @@ multidict = ">=4.0"
[metadata]
lock-version = "2.0"
python-versions = "^3.10 || ^3.11 || ^3.12"
content-hash = "3f27e66de6ea095d7574340c1233c4c5ff13409f22d3dcb2a43829acd676be30"
content-hash = "c924e8fe91a0f44fcf92650121e78da99f0e079cb9b500b0983bebc5d0dacb0c"

View file

@ -27,6 +27,7 @@ mkdocstrings = {extras = ["python"], version = "0.24.0"}
mkdocs-git-authors-plugin = "0.7.2"
mkdocs-git-revision-date-localized-plugin = "1.2.2"
mkdocs-material = {extras = ["imaging"], version = "^9.5.2"}
mkdocs-redirects = "^1.2.1"
[build-system]
requires = ["poetry-core"]

View file

@ -65,7 +65,11 @@ class RestAdapter:
try: # Deserialize JSON output to Python object, or return failed Result on exception
data_out = response.json()
except (ValueError, JSONDecodeError) as e:
raise PyZiplineError("Could not decode response from Zipline server") from e
if response.headers.get('Content-Type') == 'application/octet-stream':
# If the response contains a valid file, return success Result with file content
return Result(success=True, status_code=response.status_code, message=response.reason, data=response.content)
else:
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

View file

@ -182,7 +182,6 @@ class ZiplineApi:
Forbidden: Raised if the authenticated user is not an administrator
NotFound: Raised if the invite code does not exist
PyZiplineError: Raised if the API changes, causing a breaking change in this method
ValueError: Raised if the invite code is not provided
Returns:
Invite: An object containing the deleted invite
@ -192,8 +191,8 @@ class ZiplineApi:
return Invite(**result.data)
if result.message == 'invite not found':
raise NotFound(result.message)
if result.message == 'no code':
raise ValueError('invite code not provided')
if result.message == 'not an administrator':
raise Forbidden(result.message)
raise PyZiplineError(f"{result.status_code}: {result.message}\n{result.data}")
def get_exif(self, file_id: int) -> dict:
@ -258,6 +257,30 @@ class ZiplineApi:
if result.status_code == 401:
raise Forbidden(result.message)
def get_password_protected_file(self, file_id: int, password: str) -> bytes:
"""Get a password protected file
Args:
file_id (int): ID of the file to get
password (str): Password for the file
Raises:
Forbidden: The password is incorrect
NotFound: The file does not exist
PyZiplineError: Raised if the API changes, causing a breaking change in this method
Returns:
bytes: The file, in bytes
"""
result: Result = self._rest_adapter.get(endpoint=f"auth/image", params={'id': file_id, 'password': password})
if result.status_code == 200:
return result.data
if result.message == 'image not found':
raise NotFound(result.message)
if result.message == 'wrong password':
raise Forbidden(result.message)
raise PyZiplineError(f"{result.status_code}: {result.message}\n{result.data}")
def get_invites(self) -> list[Invite]:
"""Get a list of invites