Source code for esp.models.inventory

# -*- 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