#!/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_CREATED,
REC_FIELDNAME_CMM_TYPE,
REC_FORMAT_DATESTAMP
)
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()
[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_UPDATED,
REC_FIELDNAME_CMM_TYPE,
REC_FIELDNAME_CREATED],
localizable=False)
_id = FieldTypeFactory(MDB_FIELDNAME_ID,
localizable=False,
single_value=True)
collection = None
cmm_type = None
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 = []
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.bypass_update(self._metadata.path, self._id.path)
self.bypass_create(self._id.path)
self._metadata = self._metadata.fabricate()
self._id = self._id.fabricate()
if document_store_dictionary:
self._import_metadata(document_store_dictionary)
self._import_id(document_store_dictionary)
self.set_cmm_type()
if not document_store_dictionary:
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):
self.set_created()
self.set_updated()
return self
def _import_metadata(self, _dict):
_metadata = _dict.pop(self._metadata.get_name(), None)
if not _metadata:
return
_updated = _metadata.get(self._metadata.attr_updated.get_name(), None)
if _updated:
self.set_updated(_updated)
_created = _metadata.get(self._metadata.attr_created.get_name(), None)
if _created:
self.set_created(_created)
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 dictionary representation of record.
: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
"""
if not value:
value = datetime_now()
if not isinstance(value, datetime.datetime):
value = datestamp_to_datetime(value)
if not self._metadata.get_attribute(self._metadata.attr_updated.get_name()):
self._metadata.set_attribute(self._metadata.attr_updated.get_name(), value)
else:
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
"""
if not value:
value = datetime_now()
if not isinstance(value, datetime.datetime):
value = datestamp_to_datetime(value)
if not self._metadata.get_attribute(self._metadata.attr_created.get_name()):
self._metadata.set_attribute(self._metadata.attr_created.get_name(), value)
else:
self._metadata.attr_created.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
if not self._metadata.get_attribute(self._metadata.attr_cmm_type.get_name()):
self._metadata.set_attribute(self._metadata.attr_cmm_type.get_name(), value)
else:
self._metadata.attr_cmm_type.set_value(value)
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_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 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 not field:
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.
#: Note that two identical identifiers with same locale cannot
#: exists at same time. The latter agency will overwrite the former on update.
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')
#: 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'])
#: 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')
#: 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):
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`
"""
self.citation_requirements.add_value(value, *args)
[docs] def add_deposit_requirements(self, value, *args):
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`
"""
self.deposit_requirements.add_value(value, *args)
[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):
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`
"""
self.data_collection_copyrights.add_value(value, *args)
[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]