#!/usr/bin/env python3
# Author(s): Toni Sissala
# Copyright 2019 Finnish Social Science Data Archive FSD / University of Tampere
# Licensed under the EUPL. See LICENSE.txt for full license.
"""Models for records supported by Document Store.
Due to its schemaless design, the document store relies heavily
on these models. Use these models when building importers.
"""
import datetime
from .constants import (
MDB_FIELDNAME_ID,
REC_FIELDNAME_METADATA,
REC_FIELDNAME_UPDATED,
REC_FIELDNAME_DELETED,
REC_FIELDNAME_STATUS,
REC_FIELDNAME_SCHEMA_VERSION,
REC_FIELDNAME_CREATED,
REC_FIELDNAME_CMM_TYPE,
REC_FORMAT_DATESTAMP,
REC_STATUS_CREATED,
REC_STATUS_DELETED,
SEPARATOR_PATH
)
from .field_types import FieldTypeFactory
STUDIES = 'studies'
VARIABLES = 'variables'
QUESTIONS = 'questions'
STUDY_GROUPS = 'study_groups'
COLLECTIONS = [STUDIES, VARIABLES,
QUESTIONS, STUDY_GROUPS]
CMM_STUDY = 'study'
CMM_VARIABLE = 'variable'
CMM_QUESTION = 'question'
CMM_STUDY_GROUP = 'study_group'
CMM_TYPES = [CMM_STUDY, CMM_VARIABLE,
CMM_QUESTION, CMM_STUDY_GROUP]
[docs]def datetime_to_datestamp(_datetime):
"""Convert datetime object to datestamp string
supported by Document Store.
:param datetime: datetime to convert.
:type datetime: :class:`datetime.datetime`
:returns: converted datestamp.
:rtype: str
"""
return _datetime.strftime(REC_FORMAT_DATESTAMP)
[docs]def datestamp_to_datetime(datestamp):
"""Convert datestamp string to :class:`datetime.datetime` object.
:param datestamp: datestamp to convert.
:type datestamp: str
:returns: converted datetime.
:rtype: :class:`datetime.datetime`
"""
return datetime.datetime.strptime(datestamp, REC_FORMAT_DATESTAMP)
[docs]def datetime_now():
"""Get current datetime in supported format.
:returns: Supported datetime object representing current time.
:rtype: :obj:`datetime.datetime`
"""
return datetime.datetime.utcnow()
def to_datetime(value=None):
if not value:
value = datetime_now()
if not isinstance(value, datetime.datetime):
value = datestamp_to_datetime(value)
return value
def _convert_dict_value(_dict, key, replacement):
convertable_value = _dict.get(key)
if convertable_value is None:
return False
_dict[key] = replacement(convertable_value) if callable(replacement) else replacement
return True
[docs]def path_split(path):
"""Split path into steps.
:param str path: record attribute path.
:returns: path steps
:rtype: list
"""
return path.split(SEPARATOR_PATH)
[docs]def path_join(step, *steps):
r"""Join two or more steps into a record attribute path
by inserting a separator between the steps.
:param str step: step used to create path.
:param str *steps: steps used to create path.
:returns: Record attribute path.
:rtype: str
"""
return SEPARATOR_PATH.join((step,) + steps)
[docs]def dig_and_set(target, attr, replacement, allow_none=False):
"""Convert value corresponding to ``attr`` found from ``target`` dict / list of dicts.
Dig into ``target`` by consulting ``attr`` path. ``attr`` may either be an instance
of :obj:`Attribute` or a string. If ``replacement`` is a callable call it
with the value found from path. If it is a fixed value, insert it in place of value
found from path. If ``allow_none`` is set, raise exception if path contains no value
(is None or ``target`` has no corresponding key).
:param dict or list target: Target of replacement. May be a list or a list of dicts
:param :obj:`Attribute` or str attr: Record Class attribute or path string
:param callable or value replacement: callable returns new value or fixed value. If its a
callable, it receives old value as positional argument.
:param bool allow_none: False raises :exc:`ValueError` if ``target`` has no value for ``attr``.
:returns: None
"""
path = attr.path if hasattr(attr, 'path') else attr
# path_split always returns a list or raises exceptions.
keys = path_split(path)
final_index = len(keys) - 1
for index, key in enumerate(keys):
if not hasattr(target, 'get'):
# assume target is a list
for item in target:
dig_and_set(item, path_join(*keys[index:]), replacement, allow_none)
# dig_and_set will drill down to last key. Return afterwards.
return
if index == final_index:
# Last key. Every branch must return/raise.
converted = _convert_dict_value(target, key, replacement)
if (converted, allow_none) == (False, False):
raise ValueError("No value matching path %s in dictionary %s" % (path, target))
return
if target.get(key) is None:
continue
target = target[key]
[docs]class RecordBase:
"""Baseclass for each record.
Provides methods used to import, export, create and update
records. Dynamically fabricates each class variable of type
FieldTypeFactory into an instance variable overriding the class variable.
:note: Use this class throught subclasses only.
:param document_store_dictionary: Optional parameter for creating
a record at initialization time.
Note that this dictionary will be
iterated destructively.
:type document_store_dictionary: dict
"""
_metadata = FieldTypeFactory(REC_FIELDNAME_METADATA,
attrs=[REC_FIELDNAME_CREATED,
REC_FIELDNAME_UPDATED,
REC_FIELDNAME_DELETED,
REC_FIELDNAME_STATUS,
REC_FIELDNAME_CMM_TYPE,
REC_FIELDNAME_SCHEMA_VERSION],
localizable=False)
_id = FieldTypeFactory(MDB_FIELDNAME_ID,
localizable=False,
single_value=True)
collection = None
cmm_type = None
schema_version = '1.0'
def __init__(self, document_store_dictionary=None):
# Fabricate each class-variable of type FieldTypeFactory
# to an instance-variable of the same name.
self._update_bypass = []
self._create_bypass = []
self._fields = []
self.bypass_update(self._metadata.path, self._id.path)
self.bypass_create(self._id.path)
for att_name, att_field in self.iterate_record_fields():
_fabricated = att_field.fabricate()
self._fields.append(_fabricated)
setattr(self, att_name, _fabricated)
if document_store_dictionary and _fabricated.get_name() in document_store_dictionary:
self._import_records_to_field(document_store_dictionary, _fabricated)
self._metadata = self._metadata.fabricate()
self._id = self._id.fabricate()
if document_store_dictionary is not None:
self._import_metadata(document_store_dictionary)
self._import_id(document_store_dictionary)
else:
self._new_record()
@staticmethod
def _import_records_to_field(_dict, field):
if field.get_name() in _dict:
field.import_records(_dict[field.get_name()])
[docs] @classmethod
def get_collection(cls):
"""Get record collection.
Collection is used for queries against the Document Store.
:returns: collection of the record.
:rtype: str
"""
return cls.collection
[docs] @classmethod
def iterate_record_fields(cls):
"""Iterate class attributes used as record fields.
Iteration returns tuples: (attribute_name, attribute)
:returns: generator for iterating record fields.
"""
for att_name, att_field in cls.__dict__.items():
if hasattr(att_field, 'fabricate'):
yield att_name, att_field
def _find_field_by_name(self, name):
for field in self._fields:
if field.get_name() == name:
return field
return None
def _new_record(self):
now = to_datetime()
values = {self._metadata.attr_updated.get_name(): now,
self._metadata.attr_created.get_name(): now,
self._metadata.attr_deleted.get_name(): None,
self._metadata.attr_schema_version.get_name(): self.schema_version,
self._metadata.attr_cmm_type.get_name(): self.cmm_type,
self._metadata.attr_status.get_name(): REC_STATUS_CREATED}
self._metadata.add_value(**values)
return self
def _import_metadata(self, _dict):
"""Import existing metadata.
This method is used when client has received a record from
document store. Metadata cannot therefore be required. Also it cannot
be generated, since the architecture allows for PUTtin to docstore
without metadata (docstore handles metadata changes)
* If there is no metadata, do nothing. The client may have asked a record
without metadata and it may submit it back to docstore. The docstore is then
responsible for correct metadata.
* If there is metadata, make sure its schema is correct, and all fields are
present. If the client now submits it back to docstore, it is
responsible of changing metadata values.
* If there is metadata and its schema is invalid, raise exception.
* For migrations (schema version changes) a separate method & path is required.
The migration is a special case and it does not need to supported here.
:param dict _dict: existing metadata.
:raises: Typical datatype and value exceptions such as KeyError, ValueError and TypeError
for invalid metadata.
"""
_metadata = _dict.pop(self._metadata.get_name(), None)
if not _metadata:
return
schema_version = _metadata.pop(self._metadata.attr_schema_version.get_name())
if schema_version != self.schema_version:
raise ValueError("Invalid record schema version '%s'. Was expecting '%s'"
% (schema_version, self.schema_version))
self._metadata.attr_schema_version.set_value(schema_version)
self.set_cmm_type(_metadata.pop(self._metadata.attr_cmm_type.get_name()))
self.set_status(_metadata.pop(self._metadata.attr_status.get_name()))
self.set_created(_metadata.pop(self._metadata.attr_created.get_name()))
self.set_updated(_metadata.pop(self._metadata.attr_updated.get_name()))
deleted = _metadata.pop(self._metadata.attr_deleted.get_name())
if deleted:
self.set_deleted(deleted)
def _import_id(self, _dict):
if self._id.get_name() in _dict:
value = _dict[self._id.get_name()]
if hasattr(value, 'popitem'):
key, value = value.popitem()
self.set_id(value)
[docs] def export_dict(self, include_metadata=True, include_id=True):
"""Return JSON serializable dictionary representation of a record.
Return JSON serializable dictionary. Datetimes will be converted
to datestamps.
:param include_metadata: export includes metadata
:type include_metadata: bool
:param include_id: export includes id
:type include_id: bool
:returns: record
:rtype: dict
"""
result = {}
for field in self._fields:
result.update(field.export_dict())
if include_metadata:
result.update(self.export_metadata_dict())
if include_id and self._id.get_value():
result.update(self._id.export_dict())
return result
[docs] def set_updated(self, value=None):
"""Set updated timestamp.
Sets updated metadata attribute.
:note: The timestamp is always stored as
:class:`datetime.datetime`, but for convenience
it is accepted as a string that is formatted accordingly.
:param value: Optional timestamp to set.
:type value: :class:`datetime.datetime` or str
"""
value = to_datetime(value)
self._metadata.attr_updated.set_value(value)
return self
[docs] def set_created(self, value=None):
"""Set created timestamp.
Sets created metadata attribute.
:note: The timestamp is always stored as
:class:`datetime.datetime`, but for convenience
it is accepted as a string that is formatted accordingly.
:param value: Optional timestamp to set.
:type value: :class:`datetime.datetime` or str
"""
value = to_datetime(value)
self._metadata.attr_created.set_value(value)
return self
[docs] def set_deleted(self, value=None):
"""Set deleted timestamp.
Sets deleted metadata attribute.
:note: The timestamp is always stored as
:class:`datetime.datetime`, but for convenience
it is accepted as a string that is formatted accordingly.
:param value: Optional timestamp to set.
:type value: :class:`datetime.datetime` or str
"""
value = to_datetime(value)
self._metadata.attr_deleted.set_value(value)
return self
[docs] def set_cmm_type(self, value=None):
"""Set cmm type.
:param value: Optional type to set.
:type value: str
"""
if not value:
value = self.cmm_type
elif value != self.cmm_type:
raise ValueError("Invalid cmm type '%s'. Was expecting '%s'" % (value, self.cmm_type))
self._metadata.attr_cmm_type.set_value(value)
return self
def set_status(self, status):
val_status = (REC_STATUS_CREATED, REC_STATUS_DELETED)
if status not in (REC_STATUS_CREATED, REC_STATUS_DELETED):
raise ValueError("Invalid record status '%s'. Expected one of: %s"
% (status, ', '.join(("'%s'" % (x,) for x in val_status))))
self._metadata.attr_status.set_value(status)
return self
[docs] def set_id(self, value):
"""Set ID.
:param value: id to set.
:type value: str
"""
self._id.set_value(value)
[docs] def get_updated(self):
"""Get updated value.
:note: The timestamp is stored as a :class:`datetime.datetime`
in :attr:`_metadata.attr_updated`, but is returned
as a string datestamp when using this method.
If there is need to access the :class:`datetime.datetime`
object, use :meth:`get_value()` of the field.
:returns: updated timestamp.
:rtype: str
"""
date = self._metadata.attr_updated.get_value()
return datetime_to_datestamp(date) if date else None
[docs] def get_created(self):
"""Get created value.
:note: The timestamp is stored as a :class:`datetime.datetime`
in :attr:`_metadata.attr_created`, but is returned
as a string datestamp when using this method.
If there is need to access the :class:`datetime.datetime`
object, use :meth:`get_value()` of the field.
:returns: created timestamp.
:rtype: str
"""
date = self._metadata.attr_created.get_value()
return datetime_to_datestamp(date) if date else None
[docs] def get_deleted(self):
"""Get deleted value.
:note: The timestamp is stored as a :class:`datetime.datetime`
in :attr:`_metadata.attr_deleted`, but is returned
as a string datestamp when using this method.
If there is need to access the :class:`datetime.datetime`
object, use :meth:`get_value()` of the field.
:returns: deleted timestamp.
:rtype: str
"""
date = self._metadata.attr_deleted.get_value()
return datetime_to_datestamp(date) if date else None
[docs] def get_id(self):
"""Get record ID.
Id comes from the backend storage system.
:returns: record ID in storage.
:rtype: str or None
"""
return self._id.get_value()
[docs] def is_deleted(self):
"""Return True if record is logically deleted
:returns: True if record has been deleted.
"""
return self._metadata.attr_status.get_value() == REC_STATUS_DELETED
[docs] def bypass_update(self, *fields):
r"""Add fields to be bypassed on update operation.
:param \*fields: fieldnames to bypass.
:type \*fields: str
"""
self._update_bypass.extend(fields)
[docs] def bypass_create(self, *fields):
r"""Add fields to be bypassed on create operation.
:param \*fields: fieldnames to bypass.
:type \*fields: str
"""
self._create_bypass.extend(fields)
[docs] def updates_record(self, old_record_dict):
"""Update record by appending old values that are not present in
current record. Use old record's _id and _metadata.created if
present.
:note: parameter is a dictionary since MongoDB returns records
as JSON-like objects, which in turn are best represented
as dictionaries in python.
:param old_record_dict: Old record as a dictionary.
:type old_record_dict: dict
"""
_id = old_record_dict.pop(self._id.get_name(), None)
_metadata = old_record_dict.pop(self._metadata.get_name(), None)
if _metadata:
_created = _metadata[self._metadata.attr_created.get_name()]
for key, value in old_record_dict.items():
if key not in self._update_bypass:
field = self._find_field_by_name(key)
if field is None:
raise Exception("Unsupported field {}".format(key))
field.updates(value)
if _metadata:
self.set_created(_created)
if _id:
self.set_id(_id)
self.set_updated()
[docs] def updates(self, secondary_record):
"""Update record by appending values from secondary which are not
present in this record.
:param secondary_record: lookup values from this record.
:type secondary_record: Record instance subclassed from
:class:`RecordBase`
"""
self.updates_record(secondary_record.export_dict(
include_metadata=False, include_id=False))
[docs]class Study(RecordBase):
r"""Study record.
Derived from :class:`RecordBase`. Used to store and manipulate
Study records. Study number is a special attribute and
it cannot be updated.
All attributes of the record are declared as class
variables initiated from :class:`kuha_common.document_store.field_types.FieldTypeFactory`.
Instance methods defined in this class are used to add/set
values to record attributes. The signatures of the methods
are dynamically constructed by the definition of the
FieldTypeFactory instances. If, for example, there is a
class variable definition::
animals = FieldTypeFactory('animals', 'animal', ['color', 'weight', 'height'])
The correct method signature should be::
def add_animals(self, value, language, color=None, weight=None, height=None):
For the dynamic nature of the record-model these signatures are left
open, and python's \*args and \*\*kwargs are used instead. Note that
the field type used will raise exceptions if keyword argument
key is not found in the initial definition of the field type.
Create a new study record::
>>> study = Study()
>>> study.add_study_number(1234)
>>> study.add_study_titles('Study about animals', 'en')
>>> study.add_principal_investigators('investigator', 'en', organization='Big organization ltd.')
Import existing study record from dictionary::
>>> study_dict = {'study_number': 1234,
... 'study_titles': [{'study_title': 'Study about animals', 'language': 'en'}],
... 'principal_investigators': [{'principal_investigator': 'investigator',
... 'language': 'en', 'organization': 'Big organization ltd.'}]}
>>> study = Study(study_dict)
Iterate attributes:
>>> for pi in study.principal_investigators:
... pi.attr_organization.get_value()
...
'Big organization ltd.'
:seealso: :class:`RecordBase` and :mod:`kuha_common.document_store.field_types`
:param study_dict: Optional study record as dictionary used for constructing a record instance.
:type study_dict: dict
"""
#: Study number is used to identify a study. It must be unique within records,
#: not localizable and contain only a single value. It cannot be updated.
study_number = FieldTypeFactory('study_number',
localizable=False,
single_value=True)
#: Persistent identifiers. Multivalue-field with unique values.
persistent_identifiers = FieldTypeFactory('persistent_identifiers',
localizable=False)
#: Identifiers. Localizable field with agency-attribute.
#: This needs to be localizable for the sake of agency-attribute.
identifiers = FieldTypeFactory('identifiers',
'identifier',
'agency')
#: Study titles. Localizable, multivalue-field without attributoes.
study_titles = FieldTypeFactory('study_titles', 'study_title')
#: Document titles. Localizable, multivalue-field without attributoes.
document_titles = FieldTypeFactory('document_titles', 'document_title')
#: Parallele study titles. Localizable, multivalue-field without attributes.
parallel_titles = FieldTypeFactory('parallel_study_titles',
'parallel_study_title')
#: Pricipal investigators. Localizable, multivalue-field with organization-attribute.
principal_investigators = FieldTypeFactory('principal_investigators',
'principal_investigator',
['organization', 'external_link',
'external_link_role', 'external_link_title',
'external_link_uri'])
#: Publishers. Localizable, multivalue-field with abbreviation-attribute.
publishers = FieldTypeFactory('publishers',
'publisher',
'abbreviation')
#: Distributors. Localizable, multivalue-field with abbreviation and uri
#: attributes.
distributors = FieldTypeFactory('distributors', 'distributor',
['abbreviation', 'uri'])
#: Document URIs. Localizable, multivalue-field with location and
#: description attributes.
document_uris = FieldTypeFactory('document_uris', 'document_uri',
['location', 'description'])
#: Study URIs. Localizable, multivalue-field with location and
#: description attributes.
study_uris = FieldTypeFactory('study_uris', 'study_uri',
['location', 'description'])
#: Publication dates. Localizable, multivalue-field without attributes.
#: Note that these are treated as strings, not datetime-objects.
publication_dates = FieldTypeFactory('publication_dates', 'publication_date')
#: Publication years. Localizable, multivalue-field with
#: distribution date attribute.
publication_years = FieldTypeFactory('publication_years',
'publication_year',
'distribution_date')
#: Abstract. Localizable, multivalue-field.
abstract = FieldTypeFactory('abstracts', 'abstract')
#: Classifications. Localizable, multivalue-field with system name, uri
#: and description attributes.
classifications = FieldTypeFactory('classifications', 'classification',
['system_name', 'uri', 'description'])
#: Keywords. Localizable, multivalue-field with system name, uri
#: and description attributes.
keywords = FieldTypeFactory('keywords', 'keyword',
['system_name', 'uri', 'description'])
#: Time methods. Localizable, multivalue-field with system name, uri
#: and description attribute.
time_methods = FieldTypeFactory('time_methods', 'time_method',
['system_name', 'uri', 'description'])
#: Sampling procedures. Localizable, multivalue-field with
#: description, system name and uri attibutes.
sampling_procedures = FieldTypeFactory('sampling_procedures', 'sampling_procedure',
['description', 'system_name', 'uri'])
#: Collection modes. Localizable, multivalue-field with system name and uri
#: attritubes.
collection_modes = FieldTypeFactory('collection_modes', 'collection_mode',
['system_name', 'uri', 'description'])
#: Analysis units. Localizable, multivalue-field with system name, uri
#: and description attributes.
analysis_units = FieldTypeFactory('analysis_units', 'analysis_unit',
['system_name', 'uri', 'description'])
#: Collection periods. Localizable, multivalue-field with event-attribute.
collection_periods = FieldTypeFactory('collection_periods',
'collection_period',
'event')
#: Data kinds. Localizable, multivalue-field.
data_kinds = FieldTypeFactory('data_kinds', 'data_kind')
#: Study area countries. Localizable, multivalue-field with abbreviation attribute.
study_area_countries = FieldTypeFactory('study_area_countries',
'study_area_country',
'abbreviation')
#: Geographic coverages. Localizable, multivalue-field.
geographic_coverages = FieldTypeFactory('geographic_coverages',
'geographic_coverage')
#: Universes. Localizable, multivalue-field with included attribute.
universes = FieldTypeFactory('universes',
'universe',
'included')
#: Data access. Localizable, multivalue-field.
data_access = FieldTypeFactory('data_access', 'data_access')
#: Data access descriptions. Localizable, multivalue-field.
data_access_descriptions = FieldTypeFactory('data_access_descriptions',
'data_access_description')
#: Citation requirements. Localizable, multivalue-field.
citation_requirements = FieldTypeFactory('citation_requirements',
'citation_requirement')
#: Deposit requirements. Localizable, multivalue-field.
deposit_requirements = FieldTypeFactory('deposit_requirements',
'deposit_requirement')
#: File names. Localizable, multivalue-field.
file_names = FieldTypeFactory('file_names', 'file_name')
#: Instruments. Localizable, multivalue-field with instrument name attribute.
instruments = FieldTypeFactory('instruments',
'instrument',
'instrument_name')
#: Related publications. Localizable multivalue-field
related_publications = FieldTypeFactory('related_publications',
'related_publication',
['description', 'distribution_date', 'uri',
'identifier', 'identifier_agency'])
#: Study groups. Localizable, multivalue-field with name and description attributes.
study_groups = FieldTypeFactory('study_groups', 'study_group',
['name', 'description', 'uri'])
#: Copyrights. Localizable, multivalue-field.
copyrights = FieldTypeFactory('copyrights', 'copyright')
#: Copyrights. Localizable, multivalue-field.
data_collection_copyrights = FieldTypeFactory('data_collection_copyrights',
'data_collection_copyright')
#: Funding agencies. Localizable, multiple-field with role, agency, abbreviation,
#: description and grant number attributes.
funding_agencies = FieldTypeFactory('funding_agencies', 'funding_agency',
['role', 'abbreviation', 'description',
'grant_number'])
#: Grant numbers. Localizable, multivalue-field with agency and role attributes.
grant_numbers = FieldTypeFactory('grant_numbers', 'grant_number',
['agency', 'role'])
#: Database collection (table) for persistent storage.
collection = STUDIES
#: CMM type for Study.
cmm_type = CMM_STUDY
def __init__(self, study_dict=None):
super().__init__(study_dict)
self.bypass_update(self.study_number.get_name())
[docs] def add_study_number(self, value):
"""Add study number.
:note: despite the name, the value does not
need to be a number.
:param value: study number.
:type value: str or int
"""
self.study_number.add_value(value)
[docs] def add_persistent_identifiers(self, value):
"""Add persistent identifiers
:param value: persistent identifier
:type value: str or int
"""
self.persistent_identifiers.add_value(value)
[docs] def add_identifiers(self, value, *args, **kwargs):
r"""Add identifiers.
:param value: identifier
:type value: str or int
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.identifiers.add_value(value, *args, **kwargs)
[docs] def add_study_titles(self, value, *args, **kwargs):
r"""Add study titles.
:param value: study title.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.study_titles.add_value(value, *args, **kwargs)
[docs] def add_document_titles(self, value, *args, **kwargs):
r"""Add document titles.
:param value: document title.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.document_titles.add_value(value, *args, **kwargs)
[docs] def add_parallel_titles(self, value, *args, **kwargs):
r"""Add parallel titles.
:param value: title.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.parallel_titles.add_value(value, *args, **kwargs)
[docs] def add_principal_investigators(self, value, *args, **kwargs):
r"""Add principal investigators.
:param value: investigators.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.principal_investigators.add_value(value, *args, **kwargs)
[docs] def add_publishers(self, value, *args, **kwargs):
r"""Add publishers.
:param value: publishers.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.publishers.add_value(value, *args, **kwargs)
[docs] def add_distributors(self, value, *args, **kwargs):
r"""Add distributors.
:param value: distributor.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.distributors.add_value(value, *args, **kwargs)
[docs] def add_document_uris(self, value, *args, **kwargs):
r"""Add document URIs.
:param value: URI.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.document_uris.add_value(value, *args, **kwargs)
[docs] def add_study_uris(self, value, *args, **kwargs):
r"""Add study URIs.
:param value: URI.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.study_uris.add_value(value, *args, **kwargs)
[docs] def add_publication_dates(self, value, *args, **kwargs):
r"""Add publication dates.
:param value: date.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.publication_dates.add_value(value, *args, **kwargs)
[docs] def add_publication_years(self, value, *args, **kwargs):
r"""Add publication dates.
:param value: date.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.publication_years.add_value(value, *args, **kwargs)
[docs] def add_abstract(self, value, *args, **kwargs):
r"""Add abstract.
:param value: abstract.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.abstract.add_value(value, *args, **kwargs)
[docs] def add_classifications(self, value, *args, **kwargs):
r"""Add classifications.
:param value: classifications.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.classifications.add_value(value, *args, **kwargs)
[docs] def add_keywords(self, value, *args, **kwargs):
r"""Add keywords.
:param value: keyword.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.keywords.add_value(value, *args, **kwargs)
[docs] def add_time_methods(self, value, *args, **kwargs):
r"""Add time methods.
:param value: time method
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.time_methods.add_value(value, *args, **kwargs)
[docs] def add_sampling_procedures(self, value, *args, **kwargs):
r"""Add sampling procedures
:param value: sampling procedure
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.sampling_procedures.add_value(value, *args, **kwargs)
[docs] def add_collection_modes(self, value, *args, **kwargs):
r"""Add collection modes
:param value: collection mode
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.collection_modes.add_value(value, *args, **kwargs)
[docs] def add_analysis_units(self, value, *args, **kwargs):
r"""Add analysis units.
:param value: analysis unit.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.analysis_units.add_value(value, *args, **kwargs)
[docs] def add_collection_periods(self, value, *args, **kwargs):
r"""Add collection periods.
:param value: collection period.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.collection_periods.add_value(value, *args, **kwargs)
[docs] def add_data_kinds(self, value, *args):
r"""Add data kinds.
:param value: data kind.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.data_kinds.add_value(value, *args)
[docs] def add_study_area_countries(self, value, *args, **kwargs):
r"""Add study area countries.
:param value: country.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.study_area_countries.add_value(value, *args, **kwargs)
[docs] def add_geographic_coverages(self, value, *args):
r"""Add geographic coverages
:param value: geographic coverage
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.geographic_coverages.add_value(value, *args)
[docs] def add_universes(self, value, *args, **kwargs):
r"""Add universes.
:param value: universe.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.universes.add_value(value, *args, **kwargs)
[docs] def add_data_access(self, value, *args, **kwargs):
r"""Add data access.
:param value: data access.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.data_access.add_value(value, *args, **kwargs)
[docs] def add_data_access_descriptions(self, value, *args, **kwargs):
r"""Add data access descriptions.
:param value: access description.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.data_access_descriptions.add_value(value, *args, **kwargs)
[docs] def add_citation_requirements(self, value, *args, **kwargs):
r"""Add citation requirements.
:param value: citation requirement.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.citation_requirements.add_value(value, *args, **kwargs)
[docs] def add_deposit_requirements(self, value, *args, **kwargs):
r"""Add deposit requirements.
:param value: deposit requirement.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.deposit_requirements.add_value(value, *args, **kwargs)
[docs] def add_file_names(self, value, *args, **kwargs):
r"""Add file name.
:param value: file name.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.file_names.add_value(value, *args, **kwargs)
[docs] def add_instruments(self, value, *args, **kwargs):
r"""Add instrument.
:param value: instrument.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.instruments.add_value(value, *args, **kwargs)
[docs] def add_study_groups(self, value, *args, **kwargs):
r"""Add study group.
:param value: study group.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.study_groups.add_value(value, *args, **kwargs)
[docs] def add_copyrights(self, value, *args, **kwargs):
r"""Add copyright.
:param value: copyright.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.copyrights.add_value(value, *args, **kwargs)
[docs] def add_data_collection_copyrights(self, value, *args, **kwargs):
r"""Add data collection copyrights.
:param value: data collection copyright.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.data_collection_copyrights.add_value(value, *args, **kwargs)
[docs] def add_funding_agencies(self, value, *args, **kwargs):
r"""Add funding_agencies.
:param value: data collection copyright.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.funding_agencies.add_value(value, *args, **kwargs)
[docs] def add_grant_numbers(self, value, *args, **kwargs):
r"""Add grant numbers.
:param value: data collection copyright.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.grant_numbers.add_value(value, *args, **kwargs)
[docs] def updates(self, secondary):
"""Check that records have common unique keys.
Update record by appending values from secondary which are not
present in this record.
:param secondary: Lookup values to update from secondary record.
:type secondary: :obj:`Study`
:returns: True if record updated, False if not.
:rtype: bool
"""
found = False
if self.study_number.get_value() == secondary.study_number.get_value():
super().updates(secondary)
found = True
return found
[docs]class Variable(RecordBase):
r"""Variable record.
Derived from :class:`RecordBase`. Used to store and manipulate
variable records. Study number and variable name are special attributes and
cannot be updated.
:seealso: :class:`Study` documentation for more information.
:param variable_dict: Optional variable record as dictionary used for constructing a record instance.
:type variable_dict: dict
"""
#: Study number and variable name are used to identify a variable within variable records.
#: Their combination must be unique withing variable records, they cannot be localizable,
#: and they can only contain a single value. They also cannot be updated.
study_number = FieldTypeFactory('study_number', localizable=False, single_value=True)
#: Variable name within a study. See also :attr:`study_number`
variable_name = FieldTypeFactory('variable_name', localizable=False, single_value=True)
#: Question identifiers, if variable refers to a question.
#: Not localizable, multiple unique values.
question_identifiers = FieldTypeFactory('question_identifiers', localizable=False)
#: Variable labels. Localizable, multivalue-field.
variable_labels = FieldTypeFactory('variable_labels', 'variable_label')
#: Codelist codes. Localizable, multivalue-field with label and missing attributes.
codelist_codes = FieldTypeFactory('codelist_codes', 'codelist_code', ['label', 'missing'])
#: Database collection for persistent storage.
collection = VARIABLES
#: CMM type for variable.
cmm_type = CMM_VARIABLE
def __init__(self, variable_dict=None):
super().__init__(variable_dict)
self.bypass_update(self.study_number.get_name(),
self.variable_name.get_name())
[docs] def add_study_number(self, value):
"""Add study number.
:param value: study number.
:type value: str or int.
"""
self.study_number.add_value(value)
[docs] def add_variable_name(self, value):
"""Add variable name.
:param value: variable name.
:type value: str
"""
self.variable_name.add_value(value)
[docs] def add_question_identifiers(self, value):
"""Add question identifier
:param value: question identifier.
:type value: str or int.
"""
self.question_identifiers.add_value(value)
[docs] def add_variable_labels(self, value, *args, **kwargs):
r"""Add variable label
:param value: variable label.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.variable_labels.add_value(value, *args, **kwargs)
[docs] def add_codelist_codes(self, value, *args, **kwargs):
r"""Add codelist code
:param value: codelist code.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.codelist_codes.add_value(value, *args, **kwargs)
[docs] def updates(self, secondary):
"""Check that records have common unique keys.
Update record by appending values from secondary which are not
present in this record.
:param secondary: Lookup values to update from secondary record.
:type secondary: :obj:`Variable`
:returns: True if record updated, False if not.
:rtype: bool
"""
found = False
if self.study_number.get_value() == secondary.study_number.get_value() and\
self.variable_name.get_value() == secondary.variable_name.get_value():
super().updates(secondary)
found = True
return found
[docs]class Question(RecordBase):
r"""Question record.
Derived from :class:`RecordBase`. Used to store and manipulate
question records. :attr:`study_number` and :attr:`question_idenntifier` are special attributes and
cannot be updated.
:seealso: :class:`Study` documentation for more information.
:param question_dict: Optional question record as dictionary used for constructing a record instance.
:type question_dict: dict
"""
#: Study number and question identifier are used to identify a question. Their combination
#: must be unique withing records, they must not be localizable and they can only
#: contain a single value. They also cannot be updated.
study_number = FieldTypeFactory('study_number', localizable=False, single_value=True)
#: Question identifier within a study. See also :attr:`study_number`
question_identifier = FieldTypeFactory('question_identifier',
localizable=False,
single_value=True)
#: Variable name that specifies the variable for the question.
#: Not localizable, single value.
variable_name = FieldTypeFactory('variable_name',
localizable=False,
single_value=True)
#: Question texts. Localizable, multivalue-field.
question_texts = FieldTypeFactory('question_texts', 'question_text')
#: Research instruments. Localizable, multivalue-field.
research_instruments = FieldTypeFactory('research_instruments', 'research_instrument')
#: Codelist references. Localizable, multivalue-field.
codelist_references = FieldTypeFactory('codelist_references', 'codelist_reference')
#: Database collection for persistent storage.
collection = QUESTIONS
#: CMM type for question
cmm_type = CMM_QUESTION
def __init__(self, question_dict=None):
super().__init__(question_dict)
self.bypass_update(self.study_number.get_name(),
self.question_identifier.get_name())
[docs] def add_study_number(self, value):
"""Add study number.
:param value: study number.
:type value: str or int.
"""
self.study_number.add_value(value)
[docs] def add_question_identifier(self, value):
"""Add question identifier
:param value: question identifier.
:type value: str or int.
"""
self.question_identifier.add_value(value)
[docs] def add_variable_name(self, value):
"""Add variable name.
:param value: variable name.
:type value: str
"""
self.variable_name.add_value(value)
[docs] def add_question_texts(self, value, *args, **kwargs):
r"""Add question text
:param value: question text.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.question_texts.add_value(value, *args, **kwargs)
[docs] def add_research_instruments(self, value, *args, **kwargs):
r"""Add research instrument
:param value: research instrument.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.research_instruments.add_value(value, *args, **kwargs)
[docs] def add_codelist_references(self, value, *args, **kwargs):
r"""Add codelist reference
:param value: codelist reference
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.codelist_references.add_value(value, *args, **kwargs)
[docs] def updates(self, secondary):
"""Check that records have common unique keys.
Update record by appending values from secondary which are not
present in this record.
:param secondary: Lookup values to update from secondary record.
:type secondary: :obj:`Question`
:returns: True if record updated, False if not.
:rtype: bool
"""
found = False
if self.study_number.get_value() == secondary.study_number.get_value() and\
self.question_identifier.get_value() == secondary.question_identifier.get_value():
super().updates(secondary)
found = True
return found
[docs]class StudyGroup(RecordBase):
r"""Study group record.
Derived from :class:`RecordBase`. Used to store and manipulate study group
records. :attr:`study_group_identifier` is a arpecial attribute and
cannot be updated.
:seealso: :class:`Study` documentation for more information.
:param study_group_dict: Optional study group record as dictionary used
for constructing a record instance.
:type study_group_dict: dict
"""
#: Study group identifier. Used to identify study group.
#: Must be unique within study groups, cannot be localizable and
#: can contain only a single value. This value cannot be updated.
study_group_identifier = FieldTypeFactory('study_group_identifier',
localizable=False,
single_value=True)
#: Study group names. Localizable, multivalue-field.
study_group_names = FieldTypeFactory('study_group_names',
'study_group_name')
#: Study group descriptions. Localizable, multivalue-field.
descriptions = FieldTypeFactory('descriptions', 'description')
#: Study group URIs. Localizable, multivalue-field.
uris = FieldTypeFactory('uris', 'uri')
#: Study numbers. Multivalue-field with unique values.
study_numbers = FieldTypeFactory('study_numbers', localizable=False)
#: Database collection for persistent storage.
collection = STUDY_GROUPS
#: CMM type for study groups.
cmm_type = CMM_STUDY_GROUP
def __init__(self, study_group_dict=None):
super().__init__(study_group_dict)
self.bypass_update(self.study_group_identifier.get_name())
[docs] def add_study_group_identifier(self, value):
"""Add study group identifier.
:param value: Study group identifier.
:type value: str or int
"""
self.study_group_identifier.add_value(value)
[docs] def add_study_group_names(self, value, *args, **kwargs):
r"""Add study group names
:param value: study group name.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
:param \*\*kwargs: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.study_group_names.add_value(value, *args, **kwargs)
[docs] def add_descriptions(self, value, *args):
r"""Add study group descriptions
:param value: study group description.
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.descriptions.add_value(value, *args)
[docs] def add_uris(self, value, *args):
r"""Add study group URIs
:param value: study group URI
:type value: str
:param \*args: defined by the parameters given to
:class:`kuha_common.document_store.field_types.FieldTypeFactory`
"""
self.uris.add_value(value, *args)
[docs] def add_study_numbers(self, value):
"""Add study number.
:param value: study number.
:type value: str or int
"""
self.study_numbers.add_value(value)
[docs] def updates(self, secondary):
"""Check that records have common unique keys.
Update record by appending values from secondary which are not
present in this record.
:param secondary: Lookup values to update from secondary record.
:type secondary: :obj:`StudyGroup`
:returns: True if record updated, False if not.
:rtype: bool
"""
found = False
if self.study_group_identifier.get_value() == secondary.study_group_identifier.get_value():
super().updates(secondary)
found = True
return found
[docs]def record_factory(ds_record_dict):
"""Dynamically construct record instance
based on given document store dictionary.
Looks up the correct record by the cmm type found from
`ds_record_dict` metadata.
:param ds_record_dict: record received from Document Store.
:type ds_record_dict: dict
:returns: Record instance.
:rtype: :obj:`Study` or :obj:`Variable` or :obj:`Question`
or :obj:`StudyGroup`
"""
_cmm_type = RecordBase._metadata.attr_cmm_type.value_from_dict(ds_record_dict)
return {
CMM_STUDY: Study,
CMM_VARIABLE: Variable,
CMM_QUESTION: Question,
CMM_STUDY_GROUP: StudyGroup
}[_cmm_type](ds_record_dict)
[docs]def record_by_collection(collection):
"""Finds a record class by the given collection.
:param collection: collection of the record.
:type collection: str
:returns: record class
:rtype: :class:`Study` or :class:`Variable` or :class:`Question`
or :class:`StudyGroup`
:raises: :exc:`KeyError` if collection is not found in any record.
"""
return {
Study.collection: Study,
Variable.collection: Variable,
Question.collection: Question,
StudyGroup.collection: StudyGroup
}[collection]