# -*- coding: utf-8 -*-
#
# Report-related models.
#
# ------------------------------------------------
# imports
# -------
import json
from .. import base
from .__base__ import BaseModel, compile_report
# models
# ------
[docs]class Report(BaseModel):
"""
Object for interacting with Reports from the ESP database.
See the `Usage <./usage.html>`_ and `Examples <./examples.html>`_ pages
of the documentation for more context and comprehensive examples of
how to create and use this type of objects.
Configuration:
Create report with embedded html:
.. code-block:: yaml
name: My Report
desc: An example html report.
tags: [html, demo]
contents:
<h1>My Report</h1>
Create report with local html file:
.. code-block:: yaml
name: My Report
desc: An example html report.
tags: [html, demo]
contents: $LAB7DATA/contents/reports/my-report.html
Create applet from local html file:
.. code-block:: yaml
name: My Report
desc: An example html applet.
tags: ['esp:applet']
contents: $LAB7DATA/contents/reports/my-report.html
Configuration Notes:
* To promote an ESP report to an ESP applet, include ``esp:applet`` in the
set of report tags.
* The `contents` parameter can either take a local file or raw html contents
to use as the report data. If no file is found from the string, the data are
assumed to be raw html data to include as contents.
Examples:
.. code-block:: python
>>> from esp.models import Report
>>> report = Report('My Report')
>>> report.name, report.created_at
('My Report', '2019-06-21T16:04:01.199076Z')
>>> # show relationships
>>> report.contents
'<h1>My Report</h1>'
Arguments:
ident (str): Name or uuid for object.
"""
__api__ = "reports"
__api_cls__ = "Report"
__mutable__ = BaseModel.__mutable__ + ["parent", "elements", "report_groups", "report_type"]
__exportable__ = BaseModel.__base_exportable__ + ["report_groups", "report_type", "contents"]
[docs] @classmethod
def parse_import(cls, config, overwrite=False, allow_snapshot_uuid_remap=False):
"""
Create new object in ESP database using config file or other data.
Args:
config (str, dict, list): Config file or information to use in
creating new object.
overwrite (bool): Whether or not to delete current entry in
the ESP database.
"""
# read report contents
compiled = config.pop("compiled", False)
if config.get("elements"):
elements = [config["elements"]]
elif config.get("contents") is not None:
data = compile_report(config["contents"], compiled)
elements = [[{"type": "html", "contents": data}]]
else:
raise AssertionError("Error: html property not specified in report config!")
# normalize group spec
group = config.get("report_groups", config.get("groups", config.get("group", [])))
if not isinstance(group, (list, tuple)):
group = [group]
# create initial project
return {
"desc": config.get("desc"),
"name": config["name"],
"tags": config.get("tags", []),
"report_groups": group,
"parent": None,
"elements": elements,
"report_type": config.get("report_type", "report"),
}
@property
def contents(self):
"""
API niceness for editing internal contents, if simple html report.
"""
# TODO: need assumption guards here.
return self.elements[0][0]["contents"]
@contents.setter
def contents(self, value):
"""
API niceness for editing internal contents, if simple html report.
"""
# TODO: need assumption guards here.
self.elements[0][0]["contents"] = value
return
[docs] @classmethod
def all(cls, **kwargs):
result = base.SESSION.get("/api/reports", params={"report_type": "", "exclude_pipeline": True})
return [cls.from_data(data) for data in result.json()]
class Applet(Report):
@classmethod
def parse_import(cls, config, overwrite=False, allow_snapshot_uuid_remap=False):
ret = Report.parse_import(config, overwrite=overwrite, allow_snapshot_uuid_remap=allow_snapshot_uuid_remap)
tags = ret.setdefault("tags", [])
ret["report_type"] = "applet"
return ret
@classmethod
def all(cls, **kwargs):
result = base.SESSION.get("/api/reports", params={"report_type": "applet"})
return [cls.from_data(data) for data in result.json()]
class Query(BaseModel):
"""
Object for interacting with custom queries from the ESP database.
Represents the result of invoking the generic query handler API endpoint.
Useful for high-performance data fetches, bulk update operations, etc..
On successful query, the query result structure properties are available
as object properties, including the name of the query as defined in the
query config, the execution time, description, and parameters, and
the result data (as a json object or list of json objects, depending
on the shape of the query).
Example:
>>> from esp.models import Sample
>>> samps = Sample.create(count=4)
>>> qr = QueryResults('sample_ancestors', uuids=[x.uuid for x in samps], generation=0)
>>> len(qr.results)
4
Args:
name (str): Name of query to use.
**kwargs (dict): Parameters for query.
"""
def __init__(self, name, http_method="GET", **kwargs):
http_method = str(http_method).upper()
if http_method not in ["GET", "POST"]:
raise ValueError("http_method must be one of GET or POST but was {}".format(http_method))
self.__dict__["query_params"] = kwargs
self.__dict__["http_method"] = http_method
super(Query, self).__init__(name, **kwargs)
return
def _data_by_name(self):
"""
Overwrite _data_by_name function to issue and return query.
"""
url = "/api/v2/queries/{}".format(self.ident)
if self.http_method == "GET":
if self.query_params:
url += "?" + "&".join(
"{}={}".format(k, v if isinstance(v, str) else json.dumps(v)) for k, v in self.query_params.items()
)
result = base.SESSION.get(url).json()
else:
result = base.SESSION.post(url, json=self.query_params or {}).json()
if "error" in result:
raise AssertionError(result["error"])
return result
@classmethod
def all(cls):
"""
Override create method for instrument model.
"""
raise NotImplementedError("Querying all() query models not supported!.")
@classmethod
def create(cls, config, overwrite=False):
"""
Override create method for instrument model.
"""
raise NotImplementedError("Queries must be defined via yaml definition in deployment.")