# -*- coding: utf-8 -*-
#
# Inventory-related models.
#
# ------------------------------------------------
# imports
# -------
import datetime
import json
from ..utils import convert_date_value
try:
from urllib.parse import quote_plus
except ImportError:
from urllib import quote_plus
import logging
import dateparser
import six
from gems import cached, composite
from .. import base, utils
from .__base__ import BaseModel, LinkedModel
from .__base__ import create_and_return_uuid, export_variable_definitions, push_variable_definitions, raw
def export_meta(meta):
meta = dict(meta.json())
meta.pop("status", None)
return meta
# models
# ------
[docs]class ServiceType(BaseModel):
"""
Object for interacting with ServiceTypes 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:
Sample service type:
.. code-block:: yaml
name: Sample Service Type
desc: Service type for a demo service.
tags: [service, demo]
price_type: sample
base_price: 350
Time service type:
.. code-block:: yaml
name: Time Service Type
desc: Service type for a time service.
tags: [service, demo]
price_type: time
base_price: 350
scale_price: 25
scale_units: minutes
Configuration Notes:
* Available options for ServiceType ``price_type`` parameters are: *sample*,
*protocol*, *workflow*, *time*, or *volume*. However, only ``sample``
and ``time`` are currently supported in the UI.
* Available options for ServiceType ``scale_units`` parameters are: *hour*,
*minutes*, and *seconds*.
Arguments:
ident (str): Name or uuid for object.
"""
__api__ = "service_types"
__exportable__ = BaseModel.__base_exportable__ + [
"price_type",
"scale_price",
"scale_units",
"meta",
"customer_prices",
"base_price",
]
[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.
"""
if "customer_prices" in config:
data = config["customer_prices"]
single = not isinstance(data, (list, tuple))
if single:
data = [data]
for item in data:
if "customer_uuid" not in item:
if "customer_name" not in item:
raise AssertionError("Expected customer_name to be in customer_prices:\n{}".format(config))
uuid = Customer(item["customer_name"]).uuid
item["customer_uuid"] = uuid
return config
[docs]class Service(LinkedModel):
"""
Object for interacting with Service 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:
Simple service:
.. code-block:: yaml
name: My Service
service_type: My Service Type
customer: Lab1
Service with embedded type and customer:
.. code-block:: yaml
name: My Service
service_type:
My Service Type:
price_type: time
base_price: 350
scale_price: 25
scale_units: minutes
customer:
Lab1:
billing_address:
street1: 1234 Northword Lane
street2: Suite # 123
city: North Pole
state: North Pole
country: North Pole
postal_code: North Pole
Examples:
.. code-block:: python
>>> from esp.models import Service
>>> service = Service('My Protocol SOP')
>>> service.name, service.created_at
('My Service', '2019-06-21T16:04:01.199076Z')
>>> # show relationships
>>> service.service_type.name
'My Service Type'
>>> service.customer.name
'Customer 1'
Args:
ident (str): Name or uuid for object.
"""
__api__ = "services"
# __api_cls__ = 'Service'
__mutable__ = LinkedModel.__mutable__ + ["customer", "service_type"]
__push_format__ = {
"customer": lambda x: x.customer.uuid if x.customer is not None else None,
"service_type": lambda x: x.service_type.uuid,
}
[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.
"""
if "service_type" not in config or "customer" not in config:
raise AssertionError("`service_type` and `customer` must be in " "config for creating Service!")
config["customer"] = create_and_return_uuid(Customer, config["customer"], overwrite=overwrite)
config["service_type"] = create_and_return_uuid(ServiceType, config["service_type"], overwrite=overwrite)
return config
@cached
def service_type(self):
return ServiceType(self.data.service_type)
@cached.tag("refresh")
def customer(self):
if self.data.get("customer") is not None:
return Customer(self.data.customer)
return None
[docs]class ItemType(BaseModel):
"""
Object for interacting with ItemTypes 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:
Item type for stock reagent:
.. code-block:: yaml
name: Demo Reagent
consumption_type: stock reagent
units: uL
reorder_threshold_qty: 100
reorder_amount_qty: 50
vendors:
- Agilent
Item type for consumable:
.. code-block:: yaml
name: Demo Consumable
consumption_type: consumable
units: uL
reorder_threshold_qty: 100
reorder_amount_qty: 50
vendors:
- Agilent
Item type for kit component:
.. code-block:: yaml
name: Demo Kit Component
consumption_type: kit component
units: uL
reorder_threshold_qty: 100
reorder_amount_qty: 50
vendors:
- Agilent
Create item type with associated items:
.. code-block:: yaml
name: Consumable
consumption_type: consumable
units: uL
reorder_threshold_qty: 100
reorder_amount_qty: 50
vendors:
- Agilent
create:
- CNS001
- CNS002
Examples:
.. code-block:: python
>>> from esp.models import ItemType
>>> it = ItemType('Demo Consumable')
>>> it.name, it.created_at
('Demo Consumable', '2019-06-21T16:04:01.199076Z')
>>> # show relationships
>>> it.vendors
[<Vendor(name=Agilent)>]
Args:
ident (str): Name or uuid for object.
"""
__api__ = "item_types"
# __api_cls__ = 'ItemType'
__extend_aliases__ = True
# backwards compat from 3.0.0 -> 3.0.2.
# B/c 3.0.0 implementation of custom field support in client did not
# use the standards set by entity type and container type.
__aliases__ = {
"group": "var_group",
"ontology": "ontology_path",
}
__mutable__ = BaseModel.__mutable__ + [
"url",
"cls",
"barcode_type",
"fixed_id",
"resource_vars",
"consumption_type",
"units",
"vendor_uuids",
"reorder_threshold_qty",
"reorder_amount_qty",
"qty",
"qty_type",
"id_sequences",
]
__exportable__ = [
"name",
"desc",
"tags",
"barcode",
"barcode_type",
"reorder_threshold_qty",
"reorder_amount_qty",
"units",
"vendors",
"consumption_type",
"sequences",
"variables",
]
def _export_variables(self):
return export_variable_definitions(self.custom_fields)
__export_format__ = {
"sequences": lambda x: raw(x.sequences),
"vendors": lambda itype: [x.name for x in itype.vendors],
"variables": _export_variables,
}
__push_format__ = {
"id_sequences": lambda x: raw(x.sequences),
"resource_vars": lambda x: push_variable_definitions(raw(x.variables)),
}
[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.
"""
cls._create = config.pop("create", [])
if "vendors" in config:
vendors = [{"name": v} if isinstance(v, six.string_types) else v for v in config.pop("vendors")]
config["vendor_uuids"] = create_and_return_uuid(Vendor, vendors, overwrite=overwrite)
if not isinstance(config["vendor_uuids"], (list, tuple)):
config["vendor_uuids"] = [config["vendor_uuids"]]
# Validate consumption type. B/c the backend doesn't
if "consumption_type" in config:
if config["consumption_type"] not in ["stock reagent", "consumable", "kit component"]:
raise ValueError(
'consumption_type must be one of "stock reagent", '
'"consumable", or "kit component" (got: `{}`)'.format(config["consumption_type"])
)
sequences = config.pop("sequences", config.pop("id_sequences", None))
if sequences is not None:
# but id_sequences for PUT.
config["id_sequences"] = sequences
# Get custom fields and format
config["resource_vars"] = push_variable_definitions(config.pop("variables", []))
return config
[docs] @classmethod
def parse_response(cls, data, overwrite=False, prompt=False, allow_snapshot_uuid_remap=False):
"""
Method for parsing request and returning
:param allow_snapshot_uuid_remap:
"""
create = cls._create
del cls._create
it = cls.from_data(data)
for idx, conf in enumerate(create):
create[idx]["item_type"] = it
if len(create):
Item.create(create, overwrite=False)
return it
@property
def custom_fields(self):
return self.variables
@cached
def variables(self):
return self.resource_vars
@cached
def sequences(self):
"""
Return sequence ID used within ESP. This id is used throughout the system
in a variety of different ways, so being able to query it is useful.
"""
return list(self.data["meta"].lab7_id_sequence)
@property
def id_sequences(self):
return self.sequences
@cached.tag("refresh")
def vendors(self):
"""
Relationship to Vendor model.
"""
return [Vendor(x) for x in self.vendor_uuids]
@cached.tag("refresh")
def items(self):
"""
Relationship to Item model.
"""
result = base.SESSION.get("/api/items?item_type={}".format(quote_plus(self.name)))
return [Item.from_data(i) for i in result.json()]
[docs] def drop(self, deep=False):
"""
Issue DELETE request to remove object in ESP database
"""
if deep:
for item in self.items:
item.drop()
self.refresh()
super(ItemType, self).drop()
[docs]class Item(BaseModel):
"""
Object for interacting with ItemTypes 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:
Simple item:
.. code-block:: yaml
name: Example Item
tags: [agilent, perishable]
barcode: 45262341
barcode_type: QR
item_type: Demo Consumable
vendor: Agilent
lot_id: Lot # 1
serial_id: 12345
initial_qty: 10
expires: 12/25/2019
status: Quarantined
Item with different barcode type:
.. code-block:: yaml
name: Example Item
tags: [agilent, perishable]
barcode: 1234
barcode_type: 1D
item_type: Demo Consumable
vendor: Agilent
lot_id: ABC
Examples:
.. code-block:: python
>>> from esp.models import Item
>>> item = Item('Example Item')
>>> item.name, item.created_at
('Example Item', '2019-06-21T16:04:01.199076Z')
>>> # show relationships
>>> it.vendor
<Vendor(name=Agilent)>
>>> # useful functions
>>> item.expired()
False
>>> item.consumed()
True
>>> item.consumed()
True
>>> # interacting with item
>>> item.add_quantity(10)
>>> item.remove_quantity(5)
>>> item.empty()
>>> # interacting with variables
>>> item = Item('Example Item', params='["resource_vals"]')
>>> item.variables
>>> item.variables['Example Variable']
>>> # adding item with variables
>>> item_to_add = {
>>> 'name': 'The Name of the Item',
>>> 'description': 'Item Description',
>>> 'item_type': 'Item Type Name',
>>> 'initial_qty': 22,
>>> 'status': 'Quarantined',
>>> 'resource_vals': [{'name': 'The name of custom field', 'value': 'The Value'}]
>>> }
>>> Item.create(item_to_add)
Args:
ident (str): Name or uuid for object.
"""
__api__ = "items"
__api_cls__ = "Item"
__mutable__ = BaseModel.__mutable__ + [
"qty_note",
"add_qty",
"remove_qty",
"status",
"serial_id",
"lot_id",
"expiration_timestamp",
"vendor",
"barcode_type",
"variables",
"resource_vals",
]
__exportable__ = BaseModel.__base_exportable__ + [
"item_type",
"lot_id",
"serial_id",
"expires",
"vendor",
"status",
"initial_qty",
"qty",
"barcode",
"barcode_type",
"variables",
]
# in __base__py we have ALIASES 'value': 'default_val', 'default': 'default_val', 'default_value': 'default_val'
# I couldn't find any obvious reason for this alias 'value': 'default_val', couldn't remove this as it might break something
# Override the alias in base
# backend for Item expects resource_vals in format {'name': 'The Name', 'value': 'The Value'}
__extend_aliases__ = True
__aliases__ = {"value": "value"}
def _export_meta(self):
meta = dict(self.meta.json())
meta.pop("status", None)
return meta
__export_format__ = {
"expires": lambda x: datetime.datetime.strptime(x.expiration_timestamp.split("T")[0], "%Y-%m-%d").strftime(
"%m/%d/%Y"
)
if x.expiration_timestamp is not None
else None,
"vendor": lambda x: x.vendor.name if x.vendor else None,
"status": lambda x: x.status,
"item_type": lambda x: x.itemtype.name,
"meta": _export_meta,
# UUID is the default.
"barcode": lambda x: None if x.barcode == x.uuid else x.barcode,
# QR is the default.
"barcode_type": lambda x: None if x.barcode_type == "QR" else x.barcode_type,
"variables": lambda x: {val["name"]: val["value"] for val in x.resource_vals},
}
__pristine__ = ["variables"]
_converters = {
"date": convert_date_value,
"resource_link": utils.format_resourcelink_value,
"attachment": utils.format_attachment_value,
}
__push_format__ = {
"vendor": lambda x: x.vendor.uuid if x.vendor else None,
"resource_vals": lambda x: x._push_variables(x),
}
@classmethod
def _push_variables(cls, obj):
if isinstance(obj.item_type, str):
item_type = ItemType(obj.item_type)
else:
item_type = obj.item_type
if not hasattr(item_type, "resource_vars"):
resource_vars = []
else:
resource_vars = item_type.resource_vars
vals = raw(obj.resource_vals)
rvars = {x["name"]: x for x in resource_vars}
if isinstance(vals, list):
keys = [x["name"] for x in vals]
elif isinstance(vals, dict):
# ensure we send in the correct order.
keys = [x["name"] for x in resource_vars]
else:
raise ValueError("Unhandled resource_vals format: `{}`. Expected list or dict.".format(vals))
ret = []
for k in keys:
value = obj.variables[k]
var = rvars.get(k, {"name": k, "var_type": "string"})
if var["var_type"] in cls._converters:
value = cls._converters[var["var_type"]](value, var)
ret.append({"name": k, "value": value})
try:
obj.data.pop("resource_vals")
except KeyError:
pass
return ret
[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.
"""
# Handle item_type_uuid
if "item_type" not in config:
raise AssertionError("Expected item_type to be in config:\n{}".format(config))
if "lot_id" not in config:
# allow creating inventory items without a lot id (SDK-228).
config["lot_id"] = ""
# parse item type
item_type_name = config.pop("item_type")
if isinstance(item_type_name, BaseModel):
item_type = item_type_name
else:
item_type = ItemType(item_type_name)
if not item_type.exists():
raise AssertionError("ItemType `{}` does not exist!".format(item_type_name))
config["item_type_uuid"] = item_type.uuid
# Handle vendor_uuid, but note that vendor is optional (SDK-228)
if "vendor" not in config:
config["vendor"] = None
else:
vendor = config.pop("vendor")
if isinstance(vendor, str):
vendor = Vendor(vendor)
elif not isinstance(vendor, Vendor):
raise ValueError("vendor must be a vendor name or Vendor object")
if not vendor.exists():
raise AssertionError("Non-existent vendor: `{}`".format(vendor.name))
ok_vendors = [x.name for x in item_type.vendors]
if vendor.name not in ok_vendors:
raise AssertionError("Expected vendor to be one of {} but was `{}`".format(ok_vendors, vendor.name))
config["vendor"] = vendor.uuid
if "expires" in config or "expires_timestamp" in config:
expires = config.pop("expires", config.pop("expires_timestamp", None))
# normalize to ISO 8601.
try:
expires = dateparser.parse(expires)
except:
raise ValueError("Unhandled expires date string: `{}`".format(expires))
config["expiration_timestamp"] = expires.isoformat()
if "status" in config:
config.setdefault("meta", {})["status"] = config["status"]
return config
@cached.tag("refresh")
def vendor(self):
"""
Relationship to Vendor model.
"""
if self.data.get("vendor") is not None:
return Vendor(self.data.vendor)
return None
[docs] def add_quantity(self, amount):
"""
Add quantity to item.
"""
self.data.add_qty = amount
self.push()
del self.data["add_qty"]
return self
[docs] def remove_quantity(self, amount):
"""
Remove quantity from item.
"""
self.data.remove_qty = amount
self.push()
del self.data["remove_qty"]
return self
[docs] def empty(self):
"""
Remove all quantity from item.
"""
self.data.remove_qty = self.data.qty
self.push()
del self.data["remove_qty"]
return self
@cached.tag("refresh")
def variables(self):
rvals = self.resource_vals
# case when rvals is already a dict
if isinstance(rvals, dict) or (isinstance(rvals, composite) and rvals.meta_type == "dict"):
return dict(self.resource_vals)
return {x["name"]: x["value"] for x in self.resource_vals}
@cached.tag("refresh")
def itemtype(self):
# Note: this is the HEAD UUID, and NOT the definition UUID.
# And currently, the backend doesn't seem to support fetching
# an item_type_definition directly.
return ItemType(self.data.item_type)
@property
def expires(self):
if "expires" not in self.data or self.data["expires"] is None:
if self.data.expiration_timestamp is None:
self.data["expires"] = None
else:
try:
timestamp = self.data.expiration_timestamp
if timestamp.endswith("Z"):
timestamp = timestamp[0:-1]
self.data["expires"] = datetime.datetime.fromisoformat(timestamp)
except:
logging.warning(
"Unable to parse non-null expiration_timestamp: `%s` - expires will be None for object `%s`",
self.expiration_timestamp,
self.ident,
)
self.data["expires"] = None
return self.data["expires"]
@expires.setter
def expires(self, newval):
if isinstance(newval, str):
try:
newval = datetime.datetime.fromisoformat(newval)
except Exception as e:
raise ValueError("Unable to convert item expires value `{}` to datetime!".format(newval))
if newval is not None and not isinstance(newval, datetime.datetime):
raise ValueError("New expires value must be type None, datetime, or iso8601-formatted date/time string!")
self.data["expires"] = newval
if newval:
self.data["expiration_timestamp"] = newval.isoformat()
def _export_address(prop):
"""
billing_address:
city: city
state: state
country: country
postal_code: zip
street1: street
street2: null
tags: []
"""
def wrapper(x):
addr = getattr(x, prop)
ret = {
"street1": addr.street1,
"street2": addr.street2,
"city": addr.city,
"state": addr.state,
"country": addr.country,
"postal_code": addr.postal_code,
}
if len(addr.tags):
ret["tags"] = addr.tags.json()
return ret
return wrapper
[docs]class Customer(LinkedModel):
"""
Object for interacting with Customers 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:
Simple customer:
.. code-block:: yaml
name: Customer 1
desc: Customer 1 from the North Pole.
tags: [demo, north]
billing_address:
street1: 1234 Northword Lane
street2: Suite # 123
city: North Pole
state: North Pole
country: North Pole
postal_code: North Pole
Customer without address:
.. code-block:: yaml
name: Customer 2
desc: Customer 2 without address.
tags: [demo, north]
Customer with billing and shipping address:
.. code-block:: yaml
name: Customer 3
desc: Customer 3 from the North Pole
tags: [demo, north]
billing_address:
street1: 1234 Northword Lane
street2: Suite # 123
city: North Pole
state: North Pole
country: North Pole
postal_code: North Pole
shipping_address:
street1: 1234 Northword Lane
street2: Suite # 123
city: North Pole
state: North Pole
country: North Pole
postal_code: North Pole
Configuration Notes:
* Customer addresses are optional, and the minimal config for a customer is one
with only a specified ``name`` parameter.
Arguments:
ident (str): Name or uuid for object.
"""
__api__ = "customers"
# __api_cls__ = 'Customer'
__exportable__ = BaseModel.__mutable__ + ["billing_address", "shipping_address"]
__export_format__ = {
"billing_address": _export_address("billing_address"),
"shipping_address": _export_address("shipping_address"),
}
[docs] @classmethod
def parse_import(self, 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.
"""
if isinstance(config, six.string_types):
config = {"name": config}
return config
[docs]class Vendor(BaseModel):
"""
Object for interacting with Vendors 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:
Simple vendor:
.. code-block:: yaml
name: Vendor 1
desc: Vendor 1 from the North Pole.
tags: [demo, north]
billing_address:
street1: 1234 Northword Lane
street2: Suite # 123
city: North Pole
state: North Pole
country: North Pole
postal_code: North Pole
Vendor without address:
.. code-block:: yaml
name: Vendor 2
desc: Vendor 2 without address.
tags: [demo, north]
Vendor with billing and shipping address:
.. code-block:: yaml
name: Vendor 3
desc: Vendor 3 from the North Pole
tags: [demo, north]
billing_address:
street1: 1234 Northword Lane
street2: Suite # 123
city: North Pole
state: North Pole
country: North Pole
postal_code: North Pole
shipping_address:
street1: 1234 Northword Lane
street2: Suite # 123
city: North Pole
state: North Pole
country: North Pole
postal_code: North Pole
Configuration Notes:
* Vendor addresses are optional, and the minimal config for a customer is one
with only a specified ``name`` parameter.
Arguments:
ident (str): Name or uuid for object.
"""
__api__ = "vendors"
__exportable__ = BaseModel.__mutable__ + ["billing_address", "shipping_address"]
# __api_cls__ = 'Vendor'
def _export_address(self, address):
ret = {}
for key in ["street1", "street2", "city", "state", "country", "postal_code"]:
if address.get(key):
ret[key] = address[key]
return ret
__export_format__ = {
"billing_address": lambda x: x._export_address(x.billing_address),
"shipping_address": lambda x: x._export_address(x.shipping_address),
}
[docs] @classmethod
def parse_import(self, 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.
"""
if isinstance(config, six.string_types):
config = {"name": config}
return config