# -*- coding: utf-8 -*-
#
# Sample-related models.
#
# ------------------------------------------------
# imports
# -------
import logging
import six
from gems import cached, composite
from .. import base
from .__base__ import BaseModel
from .__base__ import yaml_str, is_uuid, create_and_return, load_config, raw
# models
# ------
[docs]class User(BaseModel):
"""
Object for interacting with users 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 user:
.. code-block:: yaml
name: New User
email: user@localhost
password: password
Create user with roles:
.. code-block:: yaml
name: New Role User
email: user-role@localhost
password: password
roles:
- Lab Manager
- Admin
Examples:
.. code-block:: python
>>> from esp.models import User
>>> user = User('New User')
>>> user.name, user.created_at
('User', '2019-06-21T16:04:01.199076Z')
>>> # show relationships
>>> user.roles
[<Role(name=Lab Manager)>, <Role(name=Admin)>]
>>> user.workgroups
[<Workgroup(name=Lab A)>]
>>> # log in as specific user
>>> User('New User').login(password='password')
>>> # get current user
>>> user = User.current()
Arguments:
ident (str): Name or uuid for object.
"""
__api__ = "users"
__api_cls__ = "User"
__defaults__ = {
"password": "password",
"roles": [],
# NOTE: THIS METADATA IS REQUIRED BY THE REQUEST
"preferences": {},
"can_create": [],
"labs": [],
"last_login_time": None,
}
__mutable__ = BaseModel.__mutable__ + [
"username",
"email",
"password",
"roles",
]
__push_format__ = {
"roles": lambda x: [{"uuid": r.uuid} for r in x.roles],
"labs": lambda x: [{"uuid": w.uuid} for w in x.workgroups],
}
__exportable__ = BaseModel.__base_exportable__ + [
"username",
"email",
"roles",
"workgroups",
]
__export_format__ = {
"roles": lambda x: [i.name for i in x.roles],
"workgroups": lambda x: [i.name for i in x.workgroups],
}
[docs] @classmethod
def parse_import(cls, config, overwrite=False, allow_snapshot_uuid_remap=False):
"""
Create new User 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.
"""
roles = create_and_return(Role, config["roles"], overwrite=overwrite)
config["roles"] = [
{
"uuid": role.uuid,
"name": role.name,
"desc": role.desc,
}
for role in roles
]
labs = create_and_return(Workgroup, config["labs"], overwrite=overwrite)
config["labs"] = [
{
"uuid": lab.uuid,
"name": lab.name,
"desc": lab.desc,
}
for lab in labs
]
return config
[docs] def drop(self, deep=False):
"""
Overwrite drop method for current user so emails are changed
on drop (allowing User re-creation to happen).
NOTE: BACKEND ARCHIVES MODEL AND MAKES IT INACCESSIBLE VIA CLIENT
THIS FUNCTIONALITY SHOULD ONLY BE USED DURING TESTING.
"""
if self.name != "system admin":
from uuid import uuid4
self.name = str(uuid4().hex) + "-archived"
self.username = self.name
self.email = self.uuid + "@localhost"
self.push()
return super(User, self).drop()
return
[docs] def login(self, password=None):
"""
Login as specific user.
"""
logging.info("Logging in as User: {}".format(self.name))
if password is None:
password = input("Password: ")
base.options(
host=base.SESSION.host,
port=base.SESSION.port,
cookies=base.SESSION.cookies,
cache=base.SESSION.cache,
username=self.username,
password=password,
)
return
[docs] @classmethod
def current(self):
"""
Return user from current session.
"""
return base.SESSION.user
@cached
def roles(self):
"""
Return roles associated with user.
"""
return [Role(x["uuid"]) for x in self.data["roles"]]
@cached
def workgroups(self):
"""
Return roles associated with user.
"""
# NOTE: BACKEND USES LABS INSTEAD OF WORKGROUPS
return [Workgroup(x["uuid"]) for x in self.data["labs"]]
[docs]class Role(BaseModel):
"""
Object for interacting with users 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 role:
.. code-block:: yaml
name: my-role
desc: My Role
permissions:
Project: [create, update]
Sample: [create, read, update, delete]
Create role with default users:
.. code-block:: yaml
name: my-role
desc: My Role
members:
- User 1
- User 2
applications:
- admin
- lims
Create role with specific app permissions:
.. code-block:: yaml
name: my-role
desc: My Role
applications:
- admin
- lims
Configuration Notes:
* Default role permissions are access to all actions for each
type of model. Permissions need to be overridden to restrict
access.
Examples:
.. code-block:: python
>>> from esp.models import Role
>>> role = Role('my-role')
>>> role.name, role.created_at
('my-role, '2019-06-21T16:04:01.199076Z')
>>> # show relationships
>>> role.users
[<User(name=User 1)>, <User(name=User 2)>]
Arguments:
ident (str): Name or uuid for object.
"""
__api__ = "roles"
__api_cls__ = "Role"
__defaults__ = {
"permissions": {},
"applications": [],
"default_application": None,
"updated_at": None,
"deleted": False,
"tags": [],
}
__mutable__ = BaseModel.__mutable__ + ["permissions", "members", "applications", "default_application", "tags"]
__push_format__ = {"members": lambda x: [{"desc": None, "name": u.name, "uuid": u.uuid} for u in x.users]}
__exportable__ = BaseModel.__base_exportable__ + ["members", "permissions", "applications", "default_application"]
__export_format__ = {
"members": lambda x: [i.name for i in x.users],
}
[docs] @classmethod
def parse_import(cls, config, overwrite=False, allow_snapshot_uuid_remap=False):
"""
Create new Role 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.
"""
for idx, member in enumerate(config.get("members", [])):
user = User(member)
if not user.exists():
raise AssertionError('Error: specified user "{}" does not exist!'.format(member))
config["members"][idx] = {"desc": None, "name": user.name, "uuid": user.uuid}
return config
[docs] def drop(self, deep=False):
"""
Overwrite drop method for current user so emails are changed
on drop (allowing User re-creation to happen).
NOTE: BACKEND ARCHIVES MODEL AND MAKES IT INACCESSIBLE VIA CLIENT
THIS FUNCTIONALITY SHOULD ONLY BE USED DURING TESTING.
"""
if self.name != "admin":
from uuid import uuid4
self.name = str(uuid4().hex) + "-archived"
self.users = []
self.push()
return super(Role, self).drop()
return
def open(self):
for key in self.data.permissions:
self.data.permissions[key] = ["create", "delete", "import", "update", "execute"]
self.data.applications = [
"admin",
"analysis",
"applets",
"builders",
"configuration",
"dashboard",
"data",
"entities",
"iam",
"ingest",
"inventory",
"lims",
"location",
"projects",
"samples",
"search",
]
self.push()
return
@cached
def users(self):
"""
Return users associated with roles.
"""
return [User.from_data(x) for x in self.data["members"]]
[docs]class Workgroup(BaseModel):
"""
Object for interacting with workgroups 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 workgroup:
.. code-block:: yaml
name: Lab A
desc: Workgroup for all users in Lab A
Create workgroup with default users:
.. code-block:: yaml
name: Lab A
desc: Workgroup for all users in Lab A
members:
- User 1
- User 2
Examples:
.. code-block:: python
>>> from esp.models import Workgroup
>>> group = Workgroup('Lab A')
>>> group.name, group.created_at
('Lab A, '2019-06-21T16:04:01.199076Z')
>>> # show relationships
>>> group.users
[<User(name=User 1)>, <User(name=User 2)>]
Arguments:
ident (str): Name or uuid for object.
"""
__api__ = "labs"
__api_cls__ = "Lab"
__defaults__ = {"members": [], "updated_at": None, "deleted": False}
__mutable__ = BaseModel.__mutable__ + [
"members",
]
__push_format__ = {"members": lambda x: [{"desc": None, "name": u.name, "uuid": u.uuid} for u in x.users]}
__exportable__ = BaseModel.__base_exportable__ + [
"members",
]
__export_format__ = {"members": lambda x: [i.name for i in x.users]}
[docs] @classmethod
def parse_import(cls, config, overwrite=False, allow_snapshot_uuid_remap=False):
"""
Create new Workgroup 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.
"""
for idx, member in enumerate(config["members"]):
user = User(member)
if not user.exists():
raise AssertionError('Error: specified user "{}" does not exist!'.format(member))
config["members"][idx] = {"desc": None, "name": user.name, "uuid": user.uuid}
return config
[docs] def drop(self, deep=False):
"""
Overwrite drop method for current user so emails are changed
on drop (allowing User re-creation to happen).
NOTE: BACKEND ARCHIVES MODEL AND MAKES IT INACCESSIBLE VIA CLIENT
THIS FUNCTIONALITY SHOULD ONLY BE USED DURING TESTING.
"""
from uuid import uuid4
self.name = str(uuid4().hex) + "-archived"
self.desc = ""
self.users = []
self.push()
return super(Workgroup, self).drop()
@cached
def users(self):
"""
Return users associated with roles.
"""
return [User.from_data(x) for x in self.data["members"]]
[docs] @classmethod
def associate_resources(cls, workgroups, resources):
"""Set the workgroup(s) to the set of workgroups for the resources."""
# TODO: CHANGE THIS TO OBJ.ASSOCIATE(RESOURCES)
group_uuids = []
resource_uuids = []
def resolve_uuid(value, type_=None):
if isinstance(value, six.string_types):
if is_uuid(value):
return value
elif type_ is None:
raise ValueError("Do not know how to handle `{}`".format(value))
else:
return type_(value).uuid
elif hasattr(value, "uuid"):
return value.uuid
else:
raise ValueError("Do not know how to handle `{}`".format(value))
for workgroup in workgroups:
group_uuids.append(resolve_uuid(workgroup, Workgroup))
for resource in resources:
resource_uuids.append(resolve_uuid(resource))
base.SESSION.post(
"/api/{}/associateresources".format(cls.__api__),
json={
"workgroups": group_uuids,
"resources": resource_uuids,
},
)
[docs]class ParamGroup(BaseModel):
"""
Object for interacting with ParamGroups 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 param group:
.. code-block:: yaml
illumina:
miseq-1: /path/to/miseq-1
miseq-2: /path/to/miseq-2
Create multiple param groups:
.. code-block:: yaml
illumina:
miseq-1: /path/to/miseq-1
miseq-2: /path/to/miseq-2
integrations:
clarity: http://clarity-url:1000
internal: http://internal-server
Examples:
.. code-block:: python
>>> from esp.models import ParamGroup
>>> pg = ParamGroup('apps')
>>> pg.name, group.created_at
('apps, '2019-06-21T16:04:01.199076Z')
>>> # show params
>>> pg.params.client
'/path/to/client/bin/esp'
Arguments:
ident (str): Name or uuid for object.
"""
__api__ = "param_groups"
__api_cls__ = "ParamGroup"
__defaults__ = {"desc": "", "params": {}}
__mutable__ = BaseModel.__mutable__ + ["params"]
@property
def __exportable__(self):
ret = list(BaseModel.__base_exportable__)
ret.extend([yaml_str(x) for x in self.params.keys()])
return ret
@property
def __export_format__(self):
formatters = {}
def formatter(name):
def _(self):
return yaml_str(self.params[name])
return _
for prop in self.__exportable__:
if prop in BaseModel.__mutable__:
continue
formatters[prop] = formatter(prop)
return formatters
[docs] @classmethod
def parse_import(cls, config, overwrite=False, allow_snapshot_uuid_remap=False):
"""
Create new Workgroup 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.
"""
# we have to suss out the configuration keys from other things,
# but we allow for the user to specify the configuration as:
# apps:
# params:
# key: value
# key: value
for key in list(config.keys()):
if key not in cls.__mutable__:
config["params"][key] = config.pop(key)
return config
[docs] def drop(self, deep=False):
"""
Overwrite drop method so 'apps' param group can't be deleted.
"""
if self.name != "apps":
return super(ParamGroup, self).drop()
return
[docs]class Config(BaseModel):
"""
Object for interacting with instance configuration within ESP.
This model allows developers to directly interact with the `/api/config`
endpoint in ESP.
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.
That is, this configuraiton is always updaitng a DB row, never truly
creating it, but uses the same interface as all other models for consistency.
This model is deprecated and will be removed in the future. Use Configuration instead.
"""
__api__ = "config"
__url__ = "/api/config"
__extend_aliases__ = False
def __init__(self, *args, **kwargs):
self.config = self._fetch_config()
super(Config, self).__init__(*args, **kwargs)
def __setattr__(self, name, value):
# completely override __setattr__ from super b/c we
# need attributes to push into _data.
if name in self.__prime_attributes__ + ["config"]:
self.__dict__[name] = value
elif name in self.__class__.__dict__:
if isinstance(self.__class__.__dict__[name], property):
self.__class__.__dict__[name].fset(self, value)
else:
self.__dict__[name] = value
else:
self._data[name] = value
def _data_by_name(self):
# Config is only allowed to write into the "augment" entry.
# other top-level entries are handled by lab7.conf.
augment = self.config.setdefault("augment", {})
if self.ident in augment and augment[self.ident] is not None:
conf = augment[self.ident]
conf["name"] = self.ident
conf["uuid"] = self.ident
return conf
return {"name": self.ident, "uuid": self.ident}
def _data_by_uuid(self):
raise AssertionError("No UUIDs for config!")
@staticmethod
def _fetch_config():
return base.SESSION.get(Config.__url__).json()
[docs] @classmethod
def all(cls, **kwargs):
config = cls._fetch_config()
ret = []
for x in config.setdefault("augment", {}):
data = config["augment"][x]
data["name"] = x
data["uuid"] = x
ret.append(Config(x, data))
return ret
[docs] @classmethod
def create(cls, config=None, overwrite=False, prompt=False, **kwargs):
"""
Create a new config 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 update a current entry in the
ESP database.
"""
# Overrides BaseModel.create.
if config is None:
config = {}
if isinstance(config, dict):
config.update(kwargs)
if not config:
raise AssertionError(
"Incorrect arguments specified to create()! Please see documentation for more information."
)
# load config. Without aliases.
data = load_config(config, {})
# manage single/multiple defs
if not isinstance(data, (list, tuple)):
data = [data]
# traverse data and create/update
created = []
for item in data:
# Config doesn't have defaults.
# break on no create or model-specific override
if item is None:
continue
# load object (or phony object) for determining if it exists
if "name" in item and item["name"] is not None:
logging.info("Creating config: {}".format(item["name"]))
config = Config(item["name"])
else:
logging.info("Creating Config Objects")
config = composite({"exists": lambda: False})
# object doesn't exist - create
if not config.exists():
result = base.SESSION.post("/api/{}".format(cls.__api__), json=item)
created.append(cls.parse_response(result.json(), overwrite=overwrite, prompt=prompt))
# object exists - update if specified
elif config.exists() and overwrite:
logging.info("Overwriting existing object {}".format(repr(config)))
name, uuid = config._data.pop("name"), config._data.pop("uuid")
config._data = composite(item)
config._data["name"] = name
config._data["uuid"] = uuid
result = config.push()
created.append(cls.parse_response(result, overwrite=overwrite, prompt=prompt))
# object exists - don't update
elif config.exists():
logging.info("Config {} already exists; will not overwrite.".format(repr(config)))
cls.parse_response(config.data.json(), overwrite=overwrite, prompt=prompt)
created.append(config)
return created if len(created) > 1 else created[0]
[docs] def push(self, dry=False):
data = self.json()
name = data.pop("name")
uuid = data.pop("uuid")
# catch issue if the user has a config value of name or uuid within this config.
if name != self.ident or uuid != self.ident:
raise AssertionError(
'"name" and "uuid" are reserved keywords for config! Expected'
" `{}`; found: name=`{}`; uuid=`{}`".format(self.ident, name, uuid)
)
self.config["augment"][self.ident] = data
base.SESSION.post(self.__url__, json=self.config)
data["name"] = name
data["uuid"] = uuid
return data
[docs] def drop(self, deep=True):
if self.ident in self.config["augment"]:
del self.config["augment"][self.ident]
base.SESSION.post(self.__url__, json=self.config)
def get(self, value, default=None):
return self.data.get(value, default)
def __contains__(self, value):
return value in self.data
def _export(self, deep, filter_empty, versioned=False):
data = self.json()
uuid = data.pop("uuid")
if data["name"] != self.ident or uuid != self.ident:
raise AssertionError(
'"name" and "uuid" are reserved keywords for config! Expected'
" `{}`; found: name=`{}`; uuid=`{}`".format(self.ident, name, uuid)
)
return data
[docs]class Configuration(BaseModel):
"""
Object for interacting with database-backed instance configuration
within ESP. This model allows developers to directly interact with
the `/api/configuration` endpoint in ESP.
Model to interact with ESP's DB-backed configuration system.
A configuration consists of two top-level dictionaries: one for
"public" configuration and one for "secret" configuration. Generally,
this model masks the distinction and makes the subkeys of both available
as top-level properties, but also supports direct access of the public and secret
dictionaries.
Examples:
.. code-block:: python
>>> from esp.models import Configuration
>>> conf = Configuration('esp') # get the ESP system config.
>>> conf.executor.num_workers
1
>>> conf = Configuration.create(name='CustomConfig', key1='value1', key2={'subkey1': 'value1.1', 'subkey2': 'value2.2'})
>>> conf.key2
{'subkey1': 'value2.1', 'subkey2': 'value2.2'}
>>> conf.key2.subkey1
'value2.1'
>>> conf.config
{'key1': 'value2', 'key2': {'subkey1': 'value2.1', 'subkey2': 'value2.2'}}
>>> conf.add_secret('secretkey', 'this value is secret')
>>> conf.secrets
{'secretkey': 'this value is secret'}
>>> conf.push()
>>> conf.refresh()
>>> conf.secrets
{'secretkey': 'XXXXXXX'}
>>> conf.secretkey
'XXXXXXX'
>>> conf.newkey = 'new value'
>>> conf.config
{'key1': 'value2', 'key2': {'subkey1': 'value2.1', 'subkey2': 'value2.2'}, 'newkey': 'new value'}
>>> conf.add_secret('newkey', 'conflicting secret')
>>> conf.newkey
'conflicting secret'
>>> conf.config.newkey
'newkey'
"""
__api__ = "configuration"
__extend_aliases__ = False
__defaults__ = {}
__pristine__ = ["config", "secrets", "schema", "locations"]
def __setattr__(self, name, value):
# completely override __setattr__ from super b/c we
# need attributes to push into config or secrets...
if name in self.__prime_attributes__ + ["config", "config_name", "desc", "name", "secrets", "uuid", "schema"]:
self.__dict__[name] = value
elif name in self.__class__.__dict__:
if isinstance(self.__class__.__dict__[name], property):
self.__class__.__dict__[name].fset(self, value)
else:
self.__dict__[name] = value
elif name in ["config_name", "config_id"]:
self.data[name] = value
elif self.data.get("secrets") and name in self.data["secrets"]:
self.data["secrets"][name] = value
elif self.data.get("schema") and name in self.data["schema"]:
self.data["schema"][name] = value
else:
self.data["config"][name] = value
def __getattr__(self, name):
if name in self.data.get("secrets", {}):
return self.data["secrets"][name]
elif name in self.data.get("config", {}):
return self.data["config"][name]
elif name in self.data.get("schema", {}):
return self.data["schema"][name]
return super(Configuration, self).__getattr__(name)
[docs] @classmethod
def parse_import(cls, config, overwrite=False, allow_snapshot_uuid_remap=False):
"""
Create new Configuration 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.
"""
ret = {}
config_name = config.pop("config_name", config.pop("name"))
ret["config_name"] = config_name
ret["name"] = config_name
# Note: the backend introduced "partial" in 2.5. Previously, partial=True was the only behavior.
# Now, the default is partial=False.
# But many of our outstanding configurations are depending on the
# older behavior, so in the _client_, we'll default to partial=True, then implementations
# can start controlling the behavior explicitly.
ret["partial"] = str(config.pop("partial", True))
ret["restart"] = str(config.pop("restart", False))
if "secrets" in config:
ret["secrets"] = config.pop("secrets")
if "schema" in config:
ret["schema"] = config.pop("schema")
if "config" in config:
ret["config"] = config.pop("config")
else:
ret["config"] = config
return ret
[docs] def add_secret(self, name, value):
"""
Add a "secret" configuration value.
Args:
name: The name
value: The real value
A secret configuration value is one where the public/REST API will return 'XXXXXX' for the value once the
value has been pushed.
"""
if "secrets" not in self.data:
self.data["secrets"] = {}
self.data["secrets"][name] = value
[docs] @classmethod
def from_data(cls, data):
cname = data.get("config_name")
return cls(ident=cname, data=data)
def _data_by_name(self):
# Config is only allowed to write into the "augment" entry.
# other top-level entries are handled by lab7.conf.
try:
data = base.SESSION.get("/api/configuration/{}".format(self.ident)).json()
except AssertionError:
return {}
data["config_name"] = self.ident
data["uuid"] = self.ident
data["name"] = self.ident
return data
def _data_by_uuid(self):
raise AssertionError("No UUIDs for config!")
def _data_by_fixed_id(self):
return self._data_by_name()
[docs] @classmethod
def all(cls, **kwargs):
"""
Fetch all active configurations.
Args:
**kwargs:
Returns:
"""
results = base.SESSION.get("/api/configuration").json()["results"]
# for now, do NOT load `esp` into the results from `all` since it is special/reserved
# and doing so menas the test infrastructure tries to delete the esp configuration, which is
# no bueno.
return [cls.from_data(x) for x in results if x["config_name"] != "esp"]
@classmethod
def _create_model(cls, item, overwrite, prompt=False):
config_name = item.pop("name", item.pop("config_name", None))
if not config_name:
raise ValueError("All configurations need a name! (missing name config: {})".format(item))
# configuration does not return the created configuration JSON. Instead, it returns # of successes and failures
# for server processes restarts if you requested a restart. We will not support that for the time being (but
# maybe in 2.4), so we can throw away the return result since a successful POST -> the configuration was created.
base.SESSION.post("/api/{}/create/{}".format(cls.__api__, config_name), json=item)
item["uuid"] = config_name
item["name"] = config_name
item["config_name"] = config_name
return cls.from_data(item)
@classmethod
def _overwrite_model(cls, obj, item, overwrite, prompt=False, allow_snapshot_uuid_remap=False):
config_name = item.pop("config_name")
data = {
"config": raw(item.get("config", {})),
}
secrets = item.get("secrets")
if secrets is not None:
data["secrets"] = raw(secrets)
schema = item.get("schema")
if schema is not None:
data["schema"] = raw(schema)
data["partial"] = item.pop("partial", "True")
data["restart"] = item.pop("restart", "False")
base.SESSION.post("/api/{}/{}".format(cls.__api__, config_name), json=data)
data["config_name"] = config_name
return cls.parse_response(data, overwrite=overwrite, prompt=prompt)
[docs] def push(self, dry=False, restart=False):
"""
Update the remote configuration with local modifications.
Args:
dry: If dry is true, returns the JSON structure that would be pushed without pushing.
Returns:
The data that was or would be pushed.
"""
config_name = self.ident
data = {
"config": raw(self.data.get("config", {})),
}
secrets = self.data.get("secrets")
if secrets is not None:
data["secrets"] = raw(secrets)
schema = self.data.get("schema")
if schema is not None:
data["schema"] = raw(schema)
if restart:
data["restart"] = "true"
if dry:
return data
base.SESSION.post("/api/{}/{}".format(self.__api__, self.ident), json=data)
return data
[docs] def get(self, name, default=None):
"""
Gets a single value from configuration, supplying a default if the key doesn't exist.
Args:
name: Name of the value to retrieve.
default: Default to use if the configuration key doesn't exist in the configuration object.
Returns:
The configured value, falling back to teh povided default if necessary
"""
return getattr(self, name, default)
def __contains__(self, value):
return hasattr(self, value)
def _export(self, deep, filter_empty, versioned=False):
data = self.json()
config = data.pop("config", {})
secrets = data.pop("secrets", {})
schema = data.pop("schema", {})
config_name = data.pop("config_name")
config["secrets"] = secrets
config["schema"] = schema
config["name"] = config_name
return config
def _export_hub(self, deep):
return self._export(deep, False)
@classmethod
def parse_import_hub(cls, data):
return cls.parse_import(data)
class Sequence(BaseModel):
"""
Object for interacting with Sequences 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 Sequence:
.. code-block:: yaml
Sequencer2:
sequences:
current_value: 1
format: MySequencer{sample_number:06}
name: Sequencer2
sequence: sequencer_for_db2
type: item
Examples:
.. code-block:: python
>>> from esp.models import Sequences
>>> sequence = Sequences('My Sequence')
>>> sequence.name, sequence.format
('My Sequence', 'MySequencer{sample_number:06}')
Arguments:
ident (str): Sequence or Format Name for object.
"""
__api__ = "sequences"
__api_cls__ = "Sequences"
__droppable__ = False
__droppable__ = False
__exportable__ = ["current_value", "id_sequences"]
@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.
"""
for id_sequence in config["id_sequences"]:
config["format"] = id_sequence.pop("format")
config["display_name"] = id_sequence.pop("name")
config["type"] = id_sequence.pop("type")
return {"sequences": [config]}
@classmethod
def from_data(cls, data):
sname = data.get("created")[0]["name"]
return cls(ident=sname, data=data)
def _data_by_name(self):
try:
data = base.SESSION.get("/api/sequences/{}".format(self.ident)).json()
except AssertionError:
return {}
data["uuid"] = self.ident
data["name"] = self.ident
return data
def _data_by_uuid(self):
raise AssertionError("No UUIDs for config!")
def _data_by_fixed_id(self):
return self._data_by_name()
def exists(self):
try:
return self.data.current_value is not None
except AttributeError:
return False
def __getattr__(self, name):
return self.data["sequences"][name]
def get(self, name, default=None):
"""
Gets a single value from sequence, supplying a default if the key doesn't exist.
Args:
name: Name of the value to retrieve.
default: Default to use if the sequence key doesn't exist in the Sequence object.
Returns:
The configured value, falling back to the provided default if necessary
"""
return getattr(self, name, default)
def __contains__(self, value):
return hasattr(self, value)
@classmethod
def all(cls, **kwargs):
result = base.SESSION.get("/api/sequences")
return [cls(ident=k, data={k: v}) for k, v in result.json()["sequences"].items()]
def _export(self, deep, filter_empty, versioned=False):
data = self.json()
return data
@classmethod
def create(cls, config=None, overwrite=False, prompt=False, object_names=None, **kwargs):
raise NotImplementedError(
"Sequence objects are currently read-only. To add new sequences, modify the appropriate Configuration."
)