Source code for pykechain.models.property_attachment
import io
import json
import os
from typing import Any, Optional
import requests
from pykechain.exceptions import APIError
from pykechain.models.property import Property
[docs]
class AttachmentProperty(Property):
"""A virtual object representing a KE-chain attachment property."""
@property
def value(self):
"""Retrieve the data value of this attachment.
Will show the filename of the attachment if there is an attachment available otherwise None
Use save_as in order to download as a file.
Example
-------
>>> file_attachment_property = project.part('Bike').property('file_attachment')
>>> if file_attachment_property.value:
... file_attachment_property.save_as('file.ext')
... else:
... print('file attachment not set, its value is None')
"""
if self.has_value():
return f"[Attachment: {self.filename}]"
else:
return None
@value.setter
def value(self, value):
if value is None:
self.clear()
else:
self.upload(data=value)
[docs]
def clear(self) -> None:
"""Clear the attachment from the attachment field.
:raises APIError: if unable to remove the attachment
"""
if self._put_value(None) is None:
self._value = None
self._json_data["value"] = None
@property
def filename(self) -> Optional[str]:
"""Filename of the attachment, without the full 'attachment' path."""
return self._value.split("/")[-1] if self.has_value() else None
[docs]
def json_load(self):
"""Download the data from the attachment and deserialise the contained json.
:return: deserialised json data as :class:`dict`
:raises APIError: When unable to retrieve the json from KE-chain
:raises JSONDecodeError: When there was a problem in deserialising the json
Example
-------
Ensure that the attachment is valid json data
>>> json_attachment = project.part('Bike').property('json_attachment')
>>> deserialised_json = json_attachment.json_load()
"""
return self._download().json()
[docs]
def upload(self, data: Any, **kwargs: Any) -> None:
"""Upload a file to the attachment property.
When providing a :class:`matplotlib.figure.Figure` object as data, the figure is uploaded as PNG.
For this, `matplotlib`_ should be installed.
:param data: File path
:type data: basestring
:raises APIError: When unable to upload the file to KE-chain
:raises OSError: When the path to the file is incorrect or file could not be found
.. _matplotlib: https://matplotlib.org/
"""
try:
import matplotlib.figure
if isinstance(data, matplotlib.figure.Figure):
self._upload_plot(data, **kwargs)
return
except ImportError:
pass
if isinstance(data, str):
with open(data, "rb") as fp:
self._upload(fp)
else:
self._upload_json(data, **kwargs)
self._value = data
[docs]
def save_as(self, filename: Optional[str] = None, **kwargs) -> None:
"""Download the attachment to a file.
:param filename: (optional) File path. If not provided, will be saved to current working dir
with `self.filename`.
:type filename: basestring or None
One can pass the `size` parameter as kwargs. See more in Enum:ImageSize or alternatively
customize the desired image size like this (width_value, height_value)
:raises APIError: When unable to download the data
:raises OSError: When unable to save the data to disk
"""
filename = filename or os.path.join(os.getcwd(), self.filename)
with open(filename, "w+b") as f:
for chunk in self._download(**kwargs):
f.write(chunk)
def _upload_json(self, content, name="data.json"):
data = (name, json.dumps(content), "application/json")
self._upload(data)
def _upload_plot(self, figure, name="plot.png"):
buffer = io.BytesIO()
figure.savefig(buffer, format="png")
data = (name, buffer.getvalue(), "image/png")
self._upload(data)
self._value = name
def _download(self, **kwargs):
url = self._client._build_url("property_download", property_id=self.id)
request_params = dict()
if kwargs:
request_params.update(**kwargs)
response = self._client._request("GET", url, params=request_params)
if response.status_code != requests.codes.ok: # pragma: no cover
raise APIError("Could not download property value.", response=response)
return response
def _upload(self, data):
url = self._client._build_url("property_upload", property_id=self.id)
response = self._client._request(
"POST",
url,
data={"part": self._json_data["part_id"]},
files={"attachment": data},
)
if response.status_code != requests.codes.ok: # pragma: no cover
raise APIError("Could not upload attachment", response=response)