Payload

Payload

Wikibase REST API payload builder for MaRDMO exports.

Provides :class:GeneratePayload, which transforms a prepared answers dict into a fully structured payload dict ready for posting to the MaRDI Portal:

• Builds Item* entries for every unique entity (labels, descriptions, aliases, statements, qualifiers).
• Builds RELATION_* entries for cross-item statements that reference other items by their temporary Item<n> placeholder.
• Builds ALIAS_* entries for alias additions to existing portal items.

Key methods: :meth:~GeneratePayload.add_data_properties, :meth:~GeneratePayload.add_check_results, :meth:~GeneratePayload.add_aliases, :meth:~GeneratePayload.build_relation_check_query.

GeneratePayload

Class to build the Payload for an Export to a Wikibase with Items, Statements, Qualifiers, and Checks.

Source code in MaRDMO/payload.py

class GeneratePayload:
    '''Class to build the Payload for an Export to a Wikibase
       with Items, Statements, Qualifiers, and Checks.'''

    def __init__(
        self,
        url: str,
        user_items: dict | None = None,
        wikibase: dict | None = None,
        dependency: dict | None = None
    ):
        '''Initialise the payload builder.

        Args:
            url:         Base URL of the target Wikibase instance (e.g.
                         ``"https://portal.mardi4nfdi.de"``).
            user_items:  Mapping of temporary ``"Item<n>"`` keys to item
                         dicts ``{ID, Name, Description}``.  Produced by
                         :func:`~MaRDMO.helpers.unique_items`.
            wikibase:    Wikibase vocabulary dicts.  Expected keys are
                         ``items``, ``properties``, ``relations``, and
                         optionally ``data_properties``.
            dependency:  Dependency graph ``{item_key: set_of_dependencies}``
                         that controls item-creation order during export.
        '''
        # Input Attributes
        self.url: str = url
        self.user_items: dict = user_items
        self.wikibase: dict =  wikibase
        self.dependency: dict= dependency
        # Working Attributes
        self.state: PayloadState = PayloadState()

    def _items_url(self):
        '''Return the Wikibase REST API endpoint URL for creating new items.'''
        return f'{self.url}/w/rest.php/wikibase/v1/entities/items'

    def _statement_url(self, item):
        '''Return the Wikibase REST API endpoint URL for adding statements to *item*.

        Args:
            item: Wikibase QID string (e.g. ``"Q42"``).
        '''
        return f'{self.url}/w/rest.php/wikibase/v1/entities/items/{item}/statements'

    def _alias_url(self, item):
        '''Return the Wikibase REST API endpoint URL for adding English aliases to *item*.

        Args:
            item: Wikibase QID string (e.g. ``"Q42"``).
        '''
        return f'{self.url}/w/rest.php/wikibase/v1/entities/items/{item}/aliases/en'

    def _build_item(self, identifier, label, description, statements = None):
        '''Build the payload dict for a new or existing Wikibase item.

        Args:
            identifier:  Existing Wikibase QID (empty string for new items).
            label:       English label string.
            description: English description string.
            statements:  Optional list of statement triples; defaults to ``[]``.

        Returns:
            Dict with keys ``id``, ``url``, ``label``, ``description``,
            and ``statements``.
        '''
        # Empty Statements if none provided
        if statements is None:
            statements = []
        if description == "No Description Provided!":
            description = ""
        # Build Item
        item = {'id': identifier,
                'url': self._items_url(),
                'label': label,
                'description':  description,
                'statements': statements}
        return item

    def _build_statement(self, identifier, content, data_type = "wikibase-item", qualifiers = None):
        '''Build the Wikibase REST API statement payload dict.

        Args:
            identifier:  Wikibase property ID string (e.g. ``"P31"``).
            content:     Statement value (QID, string, or typed literal).
            data_type:   Wikibase datatype string (default ``"wikibase-item"``).
            qualifiers:  Optional list of qualifier dicts; defaults to ``[]``.

        Returns:
            Dict in the shape expected by the Wikibase REST API
            ``POST .../statements`` endpoint.
        '''
        # Empty Qualifiers if none provided
        if qualifiers is None:
            qualifiers = []
        # Build Statement
        statement = {"statement":
                        {"property":
                            {"id": identifier,
                             "data_type": data_type},
                             "value":
                                {"type": "value",
                                 "content": content},
                             "qualifiers": qualifiers
                        }
                    }
        return statement

    def _build_alias(self, alias):
        '''Wrap *alias* in the ``{"aliases": …}`` payload dict expected by the API.

        Args:
            alias: List of alias strings.

        Returns:
            Dict ``{"aliases": alias}``.
        '''
        aliases_dict = {
          "aliases": alias
        }
        return aliases_dict

    def _normalize_aliases(self, aliases_dict: dict) -> list[str]:
        '''Convert a ``{index: alias}`` dict to a sorted, deduplicated list of strings.

        Args:
            aliases_dict: Dict mapping integer-like keys to alias strings.

        Returns:
            List of non-blank alias strings in ascending key order.
        '''
        return [
            a for _, a in sorted(aliases_dict.items())
            if isinstance(a, str) and a.strip()
        ]

    def build_relation_check_query(self):
        '''Build a SPARQL SELECT query that checks whether all RELATION entries already exist.

        Returns:
            SPARQL query string that selects one boolean variable per relation
            (``?RELATION0``, ``?RELATION1``, …).
        '''
        relation_keys = [k for k in self.state.dictionary if k.startswith('RELATION')]
        optional_blocks, bind_blocks = [], []

        for idx, key in enumerate(relation_keys):
            entry = self.state.dictionary[key]
            optional_block, bind_block = self._build_relation_block(idx, entry)
            if optional_block is None:
                continue
            optional_blocks.append(optional_block)
            bind_blocks.append(bind_block)

        query_body = '\n'.join(optional_blocks + bind_blocks)
        selectors = " ".join(f"?RELATION{idx}" for idx in range(len(relation_keys)))
        return f'\nSELECT {selectors} WHERE {{\n{query_body}\n}}'

    def _sparql_value(self, value, data_type):
        '''Format *value* as a SPARQL literal or IRI appropriate for *data_type*.

        Args:
            value:     Raw value string or QID.
            data_type: Wikibase datatype (``"wikibase-item"``, ``"string"``,
                       ``"quantity"``, ``"time"``, ``"monolingualtext"``,
                       or ``"math"``).

        Returns:
            SPARQL-formatted string (e.g. ``"wd:Q42"`` or ``"'text'"``).
        '''
        if data_type == 'wikibase-item':
            formatted_value = f'wd:{value}'
        elif data_type == 'string':
            escaped_value = value.replace("'", "\\'")
            formatted_value = f"'{escaped_value}'"
        elif data_type in ('url', 'URL'):
            formatted_value = f'<{value}>'
        elif data_type == 'quantity':
            formatted_value = f"'{value}'^^<http://www.w3.org/2001/XMLSchema#decimal>"
        elif data_type == 'time':
            formatted_value = f"'{value}'^^<http://www.w3.org/2001/XMLSchema#dateTime>"
        elif data_type == 'monolingualtext':
            escaped_value = value.replace("'", "\\'")
            formatted_value = f"'{escaped_value}'@en"
        elif data_type == 'math':
            escaped_value = value.replace("\\", "\\\\").replace("\"", "\\\"")
            formatted_value = f"'{escaped_value}'^^<http://www.w3.org/1998/Math/MathML>"
        else:
            escaped_value = value.replace("'", "\\'")
            formatted_value = f"'{escaped_value}'"
        return formatted_value

    def _build_qualifier_triples(self, qualifiers, idx):
        '''Build SPARQL triple patterns for a list of qualifier dicts.

        Args:
            qualifiers: List of qualifier dicts (each with ``property`` and
                        ``value`` sub-dicts).
            idx:        Statement index used to name the ``?statement<idx>``
                        SPARQL variable.

        Returns:
            SPARQL triple-pattern string (may be empty when *qualifiers* is empty).
        '''
        triples = ''
        for q_idx, q in enumerate(qualifiers):
            q_prop = q['property']['id']
            q_value = q['value']['content']
            q_data_type = q['property']['data_type']
            if (isinstance(q_value, str)
                    and q_value in self.state.dictionary
                    and 'id' in self.state.dictionary[q_value]):
                q_value = self.state.dictionary[q_value]['id']
                if not q_value:
                    continue
            if isinstance(q_value, dict):
                if q_data_type == 'quantity':
                    amount = q_value.get('amount', '').replace("'", "\\'")
                    triples += (
                        f'    ?statement{idx} pqv:{q_prop} ?qval{idx}_{q_idx} .\n'
                        f"    ?qval{idx}_{q_idx} wikibase:quantityAmount"
                        f" '{amount}'^^<http://www.w3.org/2001/XMLSchema#decimal> .\n"
                    )
                elif q_data_type == 'time':
                    time_str = q_value.get('time', '').replace("'", "\\'")
                    triples += (
                        f'    ?statement{idx} pqv:{q_prop} ?qval{idx}_{q_idx} .\n'
                        f"    ?qval{idx}_{q_idx} wikibase:timeValue"
                        f" '{time_str}'^^<http://www.w3.org/2001/XMLSchema#dateTime> .\n"
                    )
            else:
                triples += (
                    f'    ?statement{idx} pq:{q_prop} '
                    f'{self._sparql_value(q_value, q_data_type)} .\n'
                )
        return triples

    def _build_relation_block(self, idx, entry):
        '''Build the OPTIONAL and BIND SPARQL fragments for one RELATION entry.

        Args:
            idx:   Relation index used to name SPARQL variables.
            entry: RELATION payload dict from ``self.state.dictionary``.

        Returns:
            Tuple ``(optional_block, bind_block)`` strings, or ``(None, None)``
            when the target item is not yet in the dictionary.
        '''
        target_item_key = entry['url'].split('/')[-2]
        target_item_data = self.state.dictionary.get(target_item_key)
        if not target_item_data:
            return None, None

        target_item_id = target_item_data['id']
        statement = entry['payload']['statement']
        prop_id = statement['property']['id']
        value = statement['value']['content']
        data_type = statement['property']['data_type']
        if (isinstance(value, str)
                and value in self.state.dictionary
                and 'id' in self.state.dictionary[value]):
            value = self.state.dictionary[value]['id']
        if isinstance(value, str) and not value:
            return None, None

        subject = f'wd:{target_item_id}'
        qualifiers = statement.get('qualifiers', [])
        qual_triples = self._build_qualifier_triples(qualifiers, idx)

        if isinstance(value, dict):
            if data_type == 'quantity':
                amount = value.get('amount', '').replace("'", "\\'")
                val_lines = (
                    f'  ?statement{idx} psv:{prop_id} ?sval{idx} .\n'
                    f"  ?sval{idx} wikibase:quantityAmount"
                    f" '{amount}'^^<http://www.w3.org/2001/XMLSchema#decimal> .\n"
                )
            elif data_type == 'time':
                time_str = value.get('time', '').replace("'", "\\'")
                val_lines = (
                    f'  ?statement{idx} psv:{prop_id} ?sval{idx} .\n'
                    f"  ?sval{idx} wikibase:timeValue"
                    f" '{time_str}'^^<http://www.w3.org/2001/XMLSchema#dateTime> .\n"
                )
            else:
                return None, None
            optional = (
                f'OPTIONAL {{\n'
                f'  {subject} p:{prop_id} ?statement{idx} .\n'
                f'{val_lines}'
                f'{qual_triples if qualifiers else ""}}}'
            )
        else:
            value_str = self._sparql_value(value, data_type)
            optional = (
                f'OPTIONAL {{\n'
                f'  {subject} p:{prop_id} ?statement{idx} .\n'
                f'  ?statement{idx} ps:{prop_id} {value_str} .\n'
                f'{qual_triples if qualifiers else ""}}}'
            )

        return optional, f'BIND(BOUND(?statement{idx}) AS ?RELATION{idx})'

    def _find_key_by_values(self, id_value, name_value, description_value):
        '''Look up the ``"Item<n>"`` key matching the given ID, Name, and Description.

        Args:
            id_value:          ``ID`` field value to match.
            name_value:        ``Name`` field value to match.
            description_value: ``Description`` field value to match.

        Returns:
            Matching ``"Item<n>"`` key string, or ``None`` if not found.
        '''
        for key, values in self.user_items.items():
            if (values['ID'] == id_value and
                values['Name'] == name_value and
                values['Description'] == description_value):
                return key
        return None

    def get_dictionary(self):
        '''Return the complete payload dictionary (items, relations, and aliases).

        The returned dict maps ``"Item<n>"``, ``"RELATION<n>"``, and
        ``"ALIAS<n>"`` keys to their respective payload entries.  This is the
        top-level structure posted to the Wikibase REST API by
        :class:`~MaRDMO.oauth2.OauthProviderMixin`.
        '''
        # Get Target Dictionary
        target_dictionary = self.state.dictionary
        return target_dictionary

    def get_item_key(self, value, role='subject'):
        """Look up the ``"Item<n>"`` key for *value* and optionally set it as the current subject.

        Args:
            value: Item dict with ``ID``, ``Name``, and ``Description`` fields.
            role:  ``'subject'`` (default) stores the item as the active
                   subject for subsequent :meth:`add_answer` calls;
                   ``'object'`` just returns the key without updating state.

        Returns:
            The ``"Item<n>"`` key string for *value*.

        Raises:
            ValueError: If *value* is empty or missing ``Name``/``Description``.
        """
        if not value:
            raise ValueError("Missing Item in Statement!")
        if not value.get('Name') or not value.get('Description'):
            raise ValueError("All Items need to have a 'Name' and 'Description'!")

        item_key = self._find_key_by_values(
            value['ID'],
            value['Name'],
            value['Description'],
        )

        if role == 'subject':
            self.state.subject = value
            self.state.subject_item = item_key

        return item_key

    def set_class(self, name: str) -> None:
        '''Tag the current subject item with an entity-class label.

        Used by ``_export_*`` worker methods so that :func:`~MaRDMO.helpers.compare_items`
        can include the class in its output for grouping on the success page.

        Args:
            name: Human-readable class label (e.g. ``'Mathematical Model'``).
        '''
        if self.state.subject_item:
            self.state.dictionary[self.state.subject_item]['class'] = name

    def add_qualifier(self, identifier, data_type, content):
        '''Build a single-qualifier list for use in :meth:`add_answer`.

        Args:
            identifier: Wikibase property ID string for the qualifier (e.g. ``"P3"``).
            data_type:  Wikibase datatype string for the qualifier value.
            content:    Qualifier value (QID, string, or typed literal).

        Returns:
            Single-element list containing the qualifier dict.
        '''
        # Build Qualifer
        qualifier = [{"property":
                        {"id": identifier,
                         "data_type": data_type},
                         "value": 
                            {"type": "value",
                             "content": content}
                    }]
        return qualifier

    def add_data_properties(self, item_class):
        '''Add ``instance of`` statements for each data property selected on the current subject.

        Looks up the data-property URL → QID mapping for *item_class* and
        calls :meth:`add_answer` for every property value stored in
        ``subject["Properties"]``.

        Args:
            item_class: Entity class string passed to
                        ``self.wikibase["data_properties"]`` (e.g. ``"model"``).
        '''
        data_properties = self.wikibase['data_properties'](item_class)
        for prop in self.state.subject.get('Properties', {}).values():
            self.add_answer(
                verb=self.wikibase['properties']['instance of'],
                object_and_type=[
                    data_properties[prop],
                    'wikibase-item',
                ]
            )

    def add_check_results(self, check):
        '''Update each RELATION entry with its SPARQL existence check result.

        Args:
            check: List of SPARQL result binding dicts; the first element is
                   used (keyed by ``"RELATION<n>"``).
        '''
        relation_keys = [k for k in self.state.dictionary if k.startswith('RELATION')]
        for idx, key in enumerate(relation_keys):
            exists_key = f'RELATION{idx}'
            exists_value = check[0].get(exists_key, {}).get('value', 'false')
            self.state.dictionary[key]['exists'] = exists_value

    def check_math_relations_via_api(self, api_url):
        '''Check existence of math-datatype RELATION statements via wbgetentities.

        SPARQL returns math values as MathML, but portal stores LaTeX.  This
        method queries the Wikibase API directly to compare raw LaTeX values.

        Returns {relation_key: 'true'/'false'}.
        '''
        relation_keys = [k for k in self.state.dictionary if k.startswith('RELATION')]
        math_entries  = {}  # key → (item_qid, prop_id, latex, qualifiers)

        for key in relation_keys:
            entry     = self.state.dictionary[key]
            statement = entry['payload']['statement']
            if statement['property']['data_type'] != 'math':
                continue
            item_key = entry['url'].split('/')[-2]
            item_qid = self.state.dictionary.get(item_key, {}).get('id', '')
            if not item_qid:
                continue  # new item → statement cannot exist yet
            math_entries[key] = (
                item_qid,
                statement['property']['id'],
                statement['value']['content'],
                statement.get('qualifiers', []),
            )

        if not math_entries:
            return {}

        # Fetch claims for all relevant items
        items_needed   = set(v[0] for v in math_entries.values())
        claims_by_item = {}
        chunk_size     = 50

        for i in range(0, len(items_needed), chunk_size):
            chunk = list(items_needed)[i:i + chunk_size]
            try:
                resp = requests.get(
                    api_url,
                    params={
                        'action': 'wbgetentities',
                        'ids':    '|'.join(chunk),
                        'props':  'claims',
                        'format': 'json',
                    },
                    headers={'User-Agent': 'MaRDMO (https://zib.de; reidelbach@zib.de)'},
                    timeout=10,
                )
                resp.raise_for_status()
                for qid, entity in resp.json().get('entities', {}).items():
                    claims_by_item[qid] = entity.get('claims', {})
            except requests.exceptions.RequestException as exc:
                _logger.warning("Math relation API check failed: %s", exc)

        result = {}
        for key, (item_qid, prop_id, latex, qualifiers) in math_entries.items():
            exists = 'false'
            for claim in claims_by_item.get(item_qid, {}).get(prop_id, []):
                claim_val = claim.get('mainsnak', {}).get('datavalue', {}).get('value', '')
                if claim_val != latex:
                    continue
                if not qualifiers:
                    exists = 'true'
                    break
                # Check that all qualifiers match
                qual_match = True
                for q in qualifiers:
                    q_prop = q['property']['id']
                    q_val  = q['value']['content']
                    # Resolve temporary item key to actual QID if available
                    if q_val in self.state.dictionary:
                        q_val = self.state.dictionary[q_val].get('id', q_val)
                    api_qual_vals = {
                        c.get('datavalue', {}).get('value', {}).get('id')
                        for c in claim.get('qualifiers', {}).get(q_prop, [])
                    }
                    if q_val not in api_qual_vals:
                        qual_match = False
                        break
                if qual_match:
                    exists = 'true'
                    break
            result[key] = exists

        return result

    def add_aliases(self, aliases_dict):
        '''Add aliases for the current subject to the payload.

        For existing items (with a real QID), creates ``ALIAS`` entries for
        immediate posting.  For new items, stores the aliases on the pending
        item entry for batch creation.

        Args:
            aliases_dict: ``{index: alias_string}`` dict; does nothing when empty.
        '''
        if not aliases_dict:
            return
        aliases_list = self._normalize_aliases(aliases_dict)
        # Add Aliases
        if (
            self.state.dictionary[self.state.subject_item]['id']
        ):
            self._add_alias(
                item = self.state.subject_item,
                aliases = aliases_list
            )
        else:
            self._add_to_item_alias(
                item = self.state.subject_item,
                aliases = aliases_list
            )

    def add_answer(self, verb, object_and_type,
                   qualifier = None, subject = None):
        '''Add a single statement to the payload.

        For items that already exist on the portal (have a real QID), a
        ``RELATION`` entry is created.  For new items, the statement is
        appended to the item's ``statements`` list for batch creation.

        Args:
            verb:            Wikibase property ID string (e.g. ``"P31"``).
            object_and_type: Two-element list ``[value, datatype]`` where
                             *datatype* is a Wikibase type string such as
                             ``"wikibase-item"``, ``"string"``, or ``"math"``.
            qualifier:       Optional list of qualifier dicts built by
                             :meth:`add_qualifier`.
            subject:         ``"Item<n>"`` key of the subject item; defaults
                             to the current subject set by :meth:`get_item_key`.
        '''
        if subject is None:
            subject = self.state.subject_item
        if qualifier is None:
            qualifier = []
        # Gather Statement
        statement = {
            'property_id': verb,
            'datatype': object_and_type[1],
            'value': object_and_type[0]
        }
        # Pattern of New Item
        pattern = re.compile(r"^Item\d{10}$")
        # Add Relation
        if (
            self.state.dictionary[subject]['id']
        ):
            self._add_relation(
                item=subject,
                statement=statement,
                qualifier=qualifier
            )
        else:
            if (isinstance(statement['value'], str) and pattern.match(statement['value'])):
                self.dependency[subject].add(statement['value'])
            self._add_to_item_statement(
                item=subject,
                statement=statement,
                qualifier=qualifier
            )

    def add_answers(self, mardmo_property, wikibase_property, datatype = 'string'):
        '''Add one statement per entry in ``subject[mardmo_property]``.

        Iterates over all values stored under *mardmo_property* in the
        current subject dict and calls :meth:`add_answer` for each.

        Args:
            mardmo_property:  Key in the subject dict (e.g. ``"descriptionLong"``).
            wikibase_property: Wikibase property label looked up in
                               ``self.wikibase['properties']``.
            datatype:          Wikibase value datatype (default ``"string"``).
        '''
        for entry in self.state.subject.get(mardmo_property, {}).values():
            self.add_answer(
                verb=self.wikibase['properties'][wikibase_property],
                object_and_type=[
                    entry,
                    datatype,
                ]
            )

    def add_single_relation(
        self,
        statement,
        alt_statement = None,
        qualifier = None,
        reverse = False
    ):
        '''Add one statement per entry in ``subject[statement["relatant"]]``.

        If the relatant item exists in the payload dictionary, uses
        ``statement['relation']`` as the property; otherwise falls back to
        *alt_statement* with a string value.

        Args:
            statement:     Dict with ``relation`` (property ID) and
                           ``relatant`` (subject key) fields.
            alt_statement: Fallback dict used when the relatant is not in
                           the payload (optional).
            qualifier:     Pre-built qualifier list (optional).
            reverse:       If ``True``, the relatant becomes the subject and
                           the current item becomes the object.
        '''
        # Empty Qualifiers if none provided
        if qualifier is None:
            qualifier = []
        for entry in self.state.subject.get(statement['relatant'], {}).values():
            # Get Item Key
            entry_item = self.get_item_key(entry, 'object')
            if entry_item in self.state.dictionary:
                # Assign Object and Subject
                if reverse:
                    subject_item, object_item = entry_item, self.state.subject_item
                else:
                    subject_item, object_item = self.state.subject_item, entry_item
                # Add to Payload
                self.add_answer(
                    verb = statement['relation'],
                    object_and_type = [
                        object_item,
                        'wikibase-item',
                    ],
                    qualifier = qualifier,
                    subject = subject_item
                )
            else:
                # Add to Payload
                self.add_answer(
                    verb = alt_statement['relation'],
                    object_and_type = [
                        entry.get(alt_statement['relatant']),
                        'string',
                    ],
                    qualifier = qualifier
                )

    def add_multiple_relation(self, statement, optional_qualifier = None, reverse = False):
        '''Add one statement per (relation, relatant) pair in the current subject.

        Iterates over ``subject[statement["relation"]]`` and
        ``subject[statement["relatant"]]`` simultaneously, attaching optional
        series-ordinal, assumption, and object-role qualifiers as configured
        by *optional_qualifier*.

        Args:
            statement:          Dict with ``relation`` and ``relatant`` keys.
            optional_qualifier: List of qualifier type strings to attach; supported
                                values are ``"series ordinal"``, ``"assumes"``.
            reverse:            If ``True``, swap subject and object.
        '''

        if optional_qualifier is None:
            optional_qualifier = []

        for key, prop in self.state.subject.get(statement['relation'], {}).items():
            for key2 in self.state.subject.get(statement['relatant'], {}).get(key, {}):
                # Get Item Key
                relatant_item = self.get_item_key(
                    self.state.subject.get(statement['relatant'], {}).get(key, {}).get(key2, {}),
                    'object'
                )

                # Set Up Qualifier
                qualifier = []

                # Add Formulation and Task Order Numbers to Qualifier
                if 'series ordinal' in optional_qualifier:
                    for number in ('formulation_number', 'task_number'):
                        if self.state.subject.get(number, {}).get(key, {}):
                            qualifier = self.add_qualifier(
                                self.wikibase['properties']['series ordinal'],
                                'string', 
                                self.state.subject[number][key]
                            )

                # Add Assumptions to Qualifier
                if 'assumes' in optional_qualifier:
                    if self.state.subject.get('assumption', {}).get(key, {}):
                        for assumption in self.state.subject['assumption'][key].values():
                            assumption_item = self.get_item_key(
                                assumption,
                                'object'
                            )
                            qualifier.extend(
                                self.add_qualifier(
                                    self.wikibase['properties']['assumes'],
                                    'wikibase-item',
                                    assumption_item
                                )
                            )

                # Add Roles to Qualifier
                if len(self.wikibase['relations'][prop]) == 2:
                    if self.wikibase['relations'][prop][1] not in ('forward', 'backward'):
                        qualifier.extend(
                            self.add_qualifier(
                                self.wikibase['properties']['object has role'],
                                'wikibase-item',
                                self.wikibase['relations'][prop][1]
                            )
                        )

                # Assign Object and Subject
                if reverse or self.wikibase['relations'][prop][-1] == 'backward':
                    subject_item, object_item = relatant_item, self.state.subject_item
                else:
                    subject_item, object_item = self.state.subject_item, relatant_item

                # Add to Payload
                self.add_answer(
                    verb = self.wikibase['relations'][prop][0],
                    object_and_type = [
                        object_item,
                        'wikibase-item',
                    ],
                    qualifier = qualifier,
                    subject = subject_item
                )

    def add_in_defining_formula(self):
        '''Add ``in defining formula`` statements with ``symbol represents`` qualifiers.

        Iterates over ``subject["element"]`` entries, looks up each quantity
        item key, and adds a ``math``-typed statement linking the symbol LaTeX
        to the quantity item via a qualifier.
        '''
        for element in self.state.subject.get('element', {}).values():
            # Get Item Key
            quantity_item = self.get_item_key(
                element.get('quantity', {}),
                'object'
            )
            # Add Quantity Qualifier
            qualifier = self.add_qualifier(
                self.wikibase['properties']['symbol represents'],
                'wikibase-item',
                quantity_item
            )
            # Pattern of New Item
            pattern = re.compile(r"^Item\d{10}$")
            # Add Symbol to Payload
            if (
                self.state.dictionary[self.state.subject_item]['id']
                or self.state.subject_item == quantity_item
            ):
                self._add_relation(
                    item = self.state.subject_item,
                    statement = {
                        'property_id': self.wikibase['properties']['in defining formula'],
                        'value': element.get('symbol', ''),
                        'datatype': 'math'
                    },
                    qualifier = qualifier
                )
            else:
                if (isinstance(quantity_item, str) and pattern.match(quantity_item)):
                    self.dependency[self.state.subject_item].add(quantity_item)
                    self.add_answer(
                        verb=self.wikibase['properties']['in defining formula'],
                        object_and_type=[
                            element.get('symbol', ''),
                            'math',
                        ],
                        qualifier=qualifier
                    )

    def _add_entry(self, key, value):
        '''Insert *value* under *key* into ``self.state.dictionary``.

        Args:
            key:   Dict key (e.g. ``"Item0000000001"``).
            value: Payload entry dict to store.
        '''
        self.state.dictionary[key] = value

    def _add_to_item_alias(self, item, aliases):
        '''Store *aliases* list on the pending item entry for *item*.'''
        self.state.dictionary[item]['aliases'] = aliases

    def _add_to_item_statement(self, item, statement, qualifier=None):
        '''Append *statement* to the pending statement list of *item*.

        Used for new items that do not yet have a Wikibase QID; statements are
        batched and created together with the item.

        Args:
            item:       ``"Item<n>"`` key of the target item.
            statement:  Dict with ``property_id``, ``datatype``, and ``value``.
            qualifier:  Optional qualifier list (default ``[]``).
        '''
        if qualifier is None:
            qualifier = []
        self.state.dictionary[item]['statements'].append(
            [
                statement['property_id'],
                statement['datatype'],
                statement['value'],
                qualifier
            ]
        )

    def _add_relation(self, item, statement, qualifier=None):
        '''Create a ``RELATION<n>`` entry for a statement on an existing item.

        Args:
            item:       Wikibase QID or ``"Item<n>"`` key of the target item.
            statement:  Dict with ``property_id``, ``datatype``, and ``value``.
            qualifier:  Optional qualifier list (default ``[]``).
        '''
        if qualifier is None:
            qualifier = []
        key = f"RELATION{self.state.counter}"
        self.state.dictionary[key] = {
            'id': '',
            'url': self._statement_url(item),
            'payload': self._build_statement(
                statement['property_id'],
                statement['value'],
                statement['datatype'],
                qualifier
            )
        }
        self.state.counter += 1

    def _add_alias(self, item, aliases):
        '''Create one ``ALIAS<n>`` entry per alias for an existing item.

        Args:
            item:    Wikibase QID or ``"Item<n>"`` key of the target item.
            aliases: List of alias strings to register.
        '''
        for alias in aliases:
            key = f"ALIAS{self.state.counter}"
            self.state.dictionary[key] = {
                'id': '',
                'url': self._alias_url(item),
                'payload': self._build_alias(
                    alias = [alias]
                )
            }
            self.state.counter += 1

    def add_item_payload(self):
        '''Finalise the ``payload`` field for every pending ``Item*`` entry.

        Converts the accumulated ``statements`` list (raw ``[pid, dtype, value,
        qualifiers]`` tuples) into the Wikibase REST API item-creation format
        and stores it back in the dictionary under ``item_data["payload"]``.
        Must be called after all :meth:`add_answer` / :meth:`add_multiple_relation`
        calls and before posting.
        '''
        for item_id, item_data in self.state.dictionary.items():
            # Check if Item in Payload
            if not item_id.startswith('Item'):
                continue
            # Extract Information
            label = item_data.get("label", "")
            description = item_data.get("description", "")
            aliases = item_data.get("aliases", "")
            statements_input = item_data.get("statements", [])
            # Grouped statements by PID
            statements = {}
            for s in statements_input:
                pid, dtype, obj = s[0], s[1], s[2]
                qualifier = None
                if len(s) == 4:
                    qualifier = s[3]
                statement = {
                    "property": {"id": pid, "data_type": dtype},
                    "value": {"type": "value", "content": obj}
                }
                if qualifier:
                    statement["qualifiers"] = qualifier

                statements.setdefault(pid, []).append(statement)
            # Build payload
            payload = {
                "item": {
                    "labels": {"en": label},
                    "statements": statements
                }
            }
            if description:
                payload["item"]["descriptions"] = {"en": description}
            if aliases:
                payload["item"]["aliases"] = {"en": aliases}
            # Attach to original dict
            item_data["payload"] = payload

    def _check_mardi_and_raise(self, name: str, description: str):
        """Check if item exists in MaRDI Portal and raise error if it does."""
        mardi_identifier = query_item(name, description)
        if mardi_identifier:
            raise ValueError(
                f"An item ({mardi_identifier}) with the label '{name}' "
                f"and description '{description}' already exists on the MaRDI Portal. "
                "If you intend to use this item, please select it in the questionnaire. "
                "Otherwise, redefine it."
            )
        return mardi_identifier

    def _statement_by_id_type(self, value: dict, id_type: str):
        """Build the external-identifier statements for a user-defined item.

        Selects which identifier statements to add (Wikidata QID, ORCID iD,
        zbMath ID, or ISSN) based on *id_type* and the fields present in *value*.

        Args:
            value:   Item dict with optional keys ``ID``, ``orcid``, ``zbmath``,
                     and ``issn``.
            id_type: Source tag string (e.g. ``'wikidata'``,
                     ``'no author found'``, ``'no journal found'``).

        Returns:
            List of ``[property_id, datatype, value_str]`` statement triples.
        """
        statements = []
        if id_type == 'wikidata':
            # Add Wikidata ID
            statements.append(
                [
                    self.wikibase['properties']['Wikidata QID'],
                    'external-id',
                    value['ID'].split(':')[1]
                ]
            )
        if id_type == 'no author found':
            # Add ORCID ID Statement
            if value.get('orcid'):
                statements.append(
                    [
                        self.wikibase['properties']['ORCID iD'],
                        'external-id',
                        value['orcid']
                    ]
                )
            # Add zbMath ID Statement
            if value.get('zbmath'):
                statements.append(
                    [
                        self.wikibase['properties']['zbMATH author ID'],
                        'external-id',
                        value['zbmath']
                    ]
                )
            # If Authors has ID, add further Statements
            if statements:
                statements.append(
                    [
                        self.wikibase['properties']['instance of'],
                        'wikibase-item',
                        self.wikibase['items']['human']
                    ]
                )
                statements.append(
                    [
                        self.wikibase['properties']['MaRDI profile type'],
                        'wikibase-item',
                        self.wikibase['items']['MaRDI person profile']
                    ]
                )
        if id_type == 'no journal found':
            # Add ISSN ID Statement
            if value.get('issn'):
                statements.append(
                    [
                        self.wikibase['properties']['ISSN'],
                        'external-id',
                        value['issn']
                    ]
                )
            # Add further Statements
            statements.append(
                [
                    self.wikibase['properties']['instance of'],
                    'wikibase-item',
                    self.wikibase['items']['scientific journal']
                ]
            )
        return statements

    def process_items(self):
        """Populate the payload dictionary with an entry for every item in ``user_items``.

        Dispatches each item to a source-specific handler based on the prefix
        of its ``ID`` field (``mardi``, ``wikidata``, ``not found``,
        ``no author found``, ``no journal found``).  Existing MaRDI items are
        registered with their real QID; new items get an empty id and a
        seed list of statements.  Raises :exc:`ValueError` for Wikidata or
        user-created items whose label/description combination already exists
        on the MaRDI Portal.
        """
        handlers = {
            'mardi': lambda key, value: 
                self._add_entry(
                    key,
                    self._build_item(
                        value['ID'].split(':')[1],
                        value['Name'],
                        value['Description'],
                    )
                ),
            'wikidata': lambda key, value: (
                self._check_mardi_and_raise(
                    value['Name'],
                    value['Description']
                ),
                self._add_entry(
                    key,
                    self._build_item(
                        '',
                        value['Name'],
                        value['Description'],
                        self._statement_by_id_type(
                            value,
                            'wikidata'))
                )
            ),
            'not found': lambda key, value: (
                self._check_mardi_and_raise(
                    value['Name'],
                    value['Description']
                ),
                self._add_entry(
                    key,
                    self._build_item(
                        '',
                        value['Name'],
                        value['Description'],
                        self._statement_by_id_type(
                            value,
                            'not found'
                        )
                    )
                )
            ),
            'no author found': lambda key, value: (
                self._check_mardi_and_raise(
                    value['Name'],
                    value['Description']
                ),
                self._add_entry(
                    key,
                    self._build_item(
                        '',
                        value['Name'],
                        value['Description'],
                        self._statement_by_id_type(
                            value,
                            'no author found'
                        )
                    )
                )
            ),
            'no journal found': lambda key, value: (
                self._check_mardi_and_raise(
                    value['Name'],
                    value['Description']
                ),
                self._add_entry(
                    key,
                    self._build_item(
                        '',
                        value['Name'],
                        value['Description'],
                        self._statement_by_id_type(
                            value,
                            'no journal found'
                        )
                    )
                )
            ),
        }

        for key, value in self.user_items.items():
            if not value.get('ID'):
                continue
            for id_type, handler in handlers.items():
                if id_type in value['ID']:
                    if id_type == 'no author found':
                        if value['zbmath'] or value['orcid']:
                            handler(key, value)
                    else:
                        handler(key, value)
                    break

__init__(url, user_items=None, wikibase=None, dependency=None)

Initialise the payload builder.

Parameters:

Name	Type	Description	Default
url	str	Base URL of the target Wikibase instance (e.g. "https://portal.mardi4nfdi.de").	required
user_items	dict | None	Mapping of temporary "Item<n>" keys to item dicts {ID, Name, Description}. Produced by :func:~MaRDMO.helpers.unique_items.	None
wikibase	dict | None	Wikibase vocabulary dicts. Expected keys are items, properties, relations, and optionally data_properties.	None
dependency	dict | None	Dependency graph {item_key: set_of_dependencies} that controls item-creation order during export.	None

Source code in MaRDMO/payload.py

def __init__(
    self,
    url: str,
    user_items: dict | None = None,
    wikibase: dict | None = None,
    dependency: dict | None = None
):
    '''Initialise the payload builder.

    Args:
        url:         Base URL of the target Wikibase instance (e.g.
                     ``"https://portal.mardi4nfdi.de"``).
        user_items:  Mapping of temporary ``"Item<n>"`` keys to item
                     dicts ``{ID, Name, Description}``.  Produced by
                     :func:`~MaRDMO.helpers.unique_items`.
        wikibase:    Wikibase vocabulary dicts.  Expected keys are
                     ``items``, ``properties``, ``relations``, and
                     optionally ``data_properties``.
        dependency:  Dependency graph ``{item_key: set_of_dependencies}``
                     that controls item-creation order during export.
    '''
    # Input Attributes
    self.url: str = url
    self.user_items: dict = user_items
    self.wikibase: dict =  wikibase
    self.dependency: dict= dependency
    # Working Attributes
    self.state: PayloadState = PayloadState()

add_aliases(aliases_dict)

Add aliases for the current subject to the payload.

For existing items (with a real QID), creates ALIAS entries for immediate posting. For new items, stores the aliases on the pending item entry for batch creation.

Parameters:

Name	Type	Description	Default
aliases_dict		{index: alias_string} dict; does nothing when empty.	required

Source code in MaRDMO/payload.py

def add_aliases(self, aliases_dict):
    '''Add aliases for the current subject to the payload.

    For existing items (with a real QID), creates ``ALIAS`` entries for
    immediate posting.  For new items, stores the aliases on the pending
    item entry for batch creation.

    Args:
        aliases_dict: ``{index: alias_string}`` dict; does nothing when empty.
    '''
    if not aliases_dict:
        return
    aliases_list = self._normalize_aliases(aliases_dict)
    # Add Aliases
    if (
        self.state.dictionary[self.state.subject_item]['id']
    ):
        self._add_alias(
            item = self.state.subject_item,
            aliases = aliases_list
        )
    else:
        self._add_to_item_alias(
            item = self.state.subject_item,
            aliases = aliases_list
        )

add_answer(verb, object_and_type, qualifier=None, subject=None)

Add a single statement to the payload.

For items that already exist on the portal (have a real QID), a RELATION entry is created. For new items, the statement is appended to the item's statements list for batch creation.

Parameters:

Name	Type	Description	Default
verb		Wikibase property ID string (e.g. "P31").	required
object_and_type		Two-element list [value, datatype] where datatype is a Wikibase type string such as "wikibase-item", "string", or "math".	required
qualifier		Optional list of qualifier dicts built by :meth:add_qualifier.	None
subject		"Item<n>" key of the subject item; defaults to the current subject set by :meth:get_item_key.	None

Source code in MaRDMO/payload.py

def add_answer(self, verb, object_and_type,
               qualifier = None, subject = None):
    '''Add a single statement to the payload.

    For items that already exist on the portal (have a real QID), a
    ``RELATION`` entry is created.  For new items, the statement is
    appended to the item's ``statements`` list for batch creation.

    Args:
        verb:            Wikibase property ID string (e.g. ``"P31"``).
        object_and_type: Two-element list ``[value, datatype]`` where
                         *datatype* is a Wikibase type string such as
                         ``"wikibase-item"``, ``"string"``, or ``"math"``.
        qualifier:       Optional list of qualifier dicts built by
                         :meth:`add_qualifier`.
        subject:         ``"Item<n>"`` key of the subject item; defaults
                         to the current subject set by :meth:`get_item_key`.
    '''
    if subject is None:
        subject = self.state.subject_item
    if qualifier is None:
        qualifier = []
    # Gather Statement
    statement = {
        'property_id': verb,
        'datatype': object_and_type[1],
        'value': object_and_type[0]
    }
    # Pattern of New Item
    pattern = re.compile(r"^Item\d{10}$")
    # Add Relation
    if (
        self.state.dictionary[subject]['id']
    ):
        self._add_relation(
            item=subject,
            statement=statement,
            qualifier=qualifier
        )
    else:
        if (isinstance(statement['value'], str) and pattern.match(statement['value'])):
            self.dependency[subject].add(statement['value'])
        self._add_to_item_statement(
            item=subject,
            statement=statement,
            qualifier=qualifier
        )

add_answers(mardmo_property, wikibase_property, datatype='string')

Add one statement per entry in subject[mardmo_property].

Iterates over all values stored under mardmo_property in the current subject dict and calls :meth:add_answer for each.

Parameters:

Name	Type	Description	Default
mardmo_property		Key in the subject dict (e.g. "descriptionLong").	required
wikibase_property		Wikibase property label looked up in self.wikibase['properties'].	required
datatype		Wikibase value datatype (default "string").	'string'

Source code in MaRDMO/payload.py

def add_answers(self, mardmo_property, wikibase_property, datatype = 'string'):
    '''Add one statement per entry in ``subject[mardmo_property]``.

    Iterates over all values stored under *mardmo_property* in the
    current subject dict and calls :meth:`add_answer` for each.

    Args:
        mardmo_property:  Key in the subject dict (e.g. ``"descriptionLong"``).
        wikibase_property: Wikibase property label looked up in
                           ``self.wikibase['properties']``.
        datatype:          Wikibase value datatype (default ``"string"``).
    '''
    for entry in self.state.subject.get(mardmo_property, {}).values():
        self.add_answer(
            verb=self.wikibase['properties'][wikibase_property],
            object_and_type=[
                entry,
                datatype,
            ]
        )

add_check_results(check)

Update each RELATION entry with its SPARQL existence check result.

Parameters:

Name	Type	Description	Default
check		List of SPARQL result binding dicts; the first element is used (keyed by "RELATION<n>").	required

Source code in MaRDMO/payload.py

def add_check_results(self, check):
    '''Update each RELATION entry with its SPARQL existence check result.

    Args:
        check: List of SPARQL result binding dicts; the first element is
               used (keyed by ``"RELATION<n>"``).
    '''
    relation_keys = [k for k in self.state.dictionary if k.startswith('RELATION')]
    for idx, key in enumerate(relation_keys):
        exists_key = f'RELATION{idx}'
        exists_value = check[0].get(exists_key, {}).get('value', 'false')
        self.state.dictionary[key]['exists'] = exists_value

add_data_properties(item_class)

Add instance of statements for each data property selected on the current subject.

Looks up the data-property URL → QID mapping for item_class and calls :meth:add_answer for every property value stored in subject["Properties"].

Parameters:

Name	Type	Description	Default
item_class		Entity class string passed to self.wikibase["data_properties"] (e.g. "model").	required

Source code in MaRDMO/payload.py

def add_data_properties(self, item_class):
    '''Add ``instance of`` statements for each data property selected on the current subject.

    Looks up the data-property URL → QID mapping for *item_class* and
    calls :meth:`add_answer` for every property value stored in
    ``subject["Properties"]``.

    Args:
        item_class: Entity class string passed to
                    ``self.wikibase["data_properties"]`` (e.g. ``"model"``).
    '''
    data_properties = self.wikibase['data_properties'](item_class)
    for prop in self.state.subject.get('Properties', {}).values():
        self.add_answer(
            verb=self.wikibase['properties']['instance of'],
            object_and_type=[
                data_properties[prop],
                'wikibase-item',
            ]
        )

add_in_defining_formula()

Add in defining formula statements with symbol represents qualifiers.

Iterates over subject["element"] entries, looks up each quantity item key, and adds a math-typed statement linking the symbol LaTeX to the quantity item via a qualifier.

Source code in MaRDMO/payload.py

def add_in_defining_formula(self):
    '''Add ``in defining formula`` statements with ``symbol represents`` qualifiers.

    Iterates over ``subject["element"]`` entries, looks up each quantity
    item key, and adds a ``math``-typed statement linking the symbol LaTeX
    to the quantity item via a qualifier.
    '''
    for element in self.state.subject.get('element', {}).values():
        # Get Item Key
        quantity_item = self.get_item_key(
            element.get('quantity', {}),
            'object'
        )
        # Add Quantity Qualifier
        qualifier = self.add_qualifier(
            self.wikibase['properties']['symbol represents'],
            'wikibase-item',
            quantity_item
        )
        # Pattern of New Item
        pattern = re.compile(r"^Item\d{10}$")
        # Add Symbol to Payload
        if (
            self.state.dictionary[self.state.subject_item]['id']
            or self.state.subject_item == quantity_item
        ):
            self._add_relation(
                item = self.state.subject_item,
                statement = {
                    'property_id': self.wikibase['properties']['in defining formula'],
                    'value': element.get('symbol', ''),
                    'datatype': 'math'
                },
                qualifier = qualifier
            )
        else:
            if (isinstance(quantity_item, str) and pattern.match(quantity_item)):
                self.dependency[self.state.subject_item].add(quantity_item)
                self.add_answer(
                    verb=self.wikibase['properties']['in defining formula'],
                    object_and_type=[
                        element.get('symbol', ''),
                        'math',
                    ],
                    qualifier=qualifier
                )

add_item_payload()

Finalise the payload field for every pending Item* entry.

Converts the accumulated statements list (raw [pid, dtype, value, qualifiers] tuples) into the Wikibase REST API item-creation format and stores it back in the dictionary under item_data["payload"]. Must be called after all :meth:add_answer / :meth:add_multiple_relation calls and before posting.

Source code in MaRDMO/payload.py

def add_item_payload(self):
    '''Finalise the ``payload`` field for every pending ``Item*`` entry.

    Converts the accumulated ``statements`` list (raw ``[pid, dtype, value,
    qualifiers]`` tuples) into the Wikibase REST API item-creation format
    and stores it back in the dictionary under ``item_data["payload"]``.
    Must be called after all :meth:`add_answer` / :meth:`add_multiple_relation`
    calls and before posting.
    '''
    for item_id, item_data in self.state.dictionary.items():
        # Check if Item in Payload
        if not item_id.startswith('Item'):
            continue
        # Extract Information
        label = item_data.get("label", "")
        description = item_data.get("description", "")
        aliases = item_data.get("aliases", "")
        statements_input = item_data.get("statements", [])
        # Grouped statements by PID
        statements = {}
        for s in statements_input:
            pid, dtype, obj = s[0], s[1], s[2]
            qualifier = None
            if len(s) == 4:
                qualifier = s[3]
            statement = {
                "property": {"id": pid, "data_type": dtype},
                "value": {"type": "value", "content": obj}
            }
            if qualifier:
                statement["qualifiers"] = qualifier

            statements.setdefault(pid, []).append(statement)
        # Build payload
        payload = {
            "item": {
                "labels": {"en": label},
                "statements": statements
            }
        }
        if description:
            payload["item"]["descriptions"] = {"en": description}
        if aliases:
            payload["item"]["aliases"] = {"en": aliases}
        # Attach to original dict
        item_data["payload"] = payload

add_multiple_relation(statement, optional_qualifier=None, reverse=False)

Add one statement per (relation, relatant) pair in the current subject.

Iterates over subject[statement["relation"]] and subject[statement["relatant"]] simultaneously, attaching optional series-ordinal, assumption, and object-role qualifiers as configured by optional_qualifier.

Parameters:

Name	Type	Description	Default
statement		Dict with relation and relatant keys.	required
optional_qualifier		List of qualifier type strings to attach; supported values are "series ordinal", "assumes".	None
reverse		If True, swap subject and object.	False

Source code in MaRDMO/payload.py

def add_multiple_relation(self, statement, optional_qualifier = None, reverse = False):
    '''Add one statement per (relation, relatant) pair in the current subject.

    Iterates over ``subject[statement["relation"]]`` and
    ``subject[statement["relatant"]]`` simultaneously, attaching optional
    series-ordinal, assumption, and object-role qualifiers as configured
    by *optional_qualifier*.

    Args:
        statement:          Dict with ``relation`` and ``relatant`` keys.
        optional_qualifier: List of qualifier type strings to attach; supported
                            values are ``"series ordinal"``, ``"assumes"``.
        reverse:            If ``True``, swap subject and object.
    '''

    if optional_qualifier is None:
        optional_qualifier = []

    for key, prop in self.state.subject.get(statement['relation'], {}).items():
        for key2 in self.state.subject.get(statement['relatant'], {}).get(key, {}):
            # Get Item Key
            relatant_item = self.get_item_key(
                self.state.subject.get(statement['relatant'], {}).get(key, {}).get(key2, {}),
                'object'
            )

            # Set Up Qualifier
            qualifier = []

            # Add Formulation and Task Order Numbers to Qualifier
            if 'series ordinal' in optional_qualifier:
                for number in ('formulation_number', 'task_number'):
                    if self.state.subject.get(number, {}).get(key, {}):
                        qualifier = self.add_qualifier(
                            self.wikibase['properties']['series ordinal'],
                            'string', 
                            self.state.subject[number][key]
                        )

            # Add Assumptions to Qualifier
            if 'assumes' in optional_qualifier:
                if self.state.subject.get('assumption', {}).get(key, {}):
                    for assumption in self.state.subject['assumption'][key].values():
                        assumption_item = self.get_item_key(
                            assumption,
                            'object'
                        )
                        qualifier.extend(
                            self.add_qualifier(
                                self.wikibase['properties']['assumes'],
                                'wikibase-item',
                                assumption_item
                            )
                        )

            # Add Roles to Qualifier
            if len(self.wikibase['relations'][prop]) == 2:
                if self.wikibase['relations'][prop][1] not in ('forward', 'backward'):
                    qualifier.extend(
                        self.add_qualifier(
                            self.wikibase['properties']['object has role'],
                            'wikibase-item',
                            self.wikibase['relations'][prop][1]
                        )
                    )

            # Assign Object and Subject
            if reverse or self.wikibase['relations'][prop][-1] == 'backward':
                subject_item, object_item = relatant_item, self.state.subject_item
            else:
                subject_item, object_item = self.state.subject_item, relatant_item

            # Add to Payload
            self.add_answer(
                verb = self.wikibase['relations'][prop][0],
                object_and_type = [
                    object_item,
                    'wikibase-item',
                ],
                qualifier = qualifier,
                subject = subject_item
            )

add_qualifier(identifier, data_type, content)

Build a single-qualifier list for use in :meth:add_answer.

Parameters:

Name	Type	Description	Default
identifier		Wikibase property ID string for the qualifier (e.g. "P3").	required
data_type		Wikibase datatype string for the qualifier value.	required
content		Qualifier value (QID, string, or typed literal).	required

Returns:

Type	Description
	Single-element list containing the qualifier dict.

Source code in MaRDMO/payload.py

def add_qualifier(self, identifier, data_type, content):
    '''Build a single-qualifier list for use in :meth:`add_answer`.

    Args:
        identifier: Wikibase property ID string for the qualifier (e.g. ``"P3"``).
        data_type:  Wikibase datatype string for the qualifier value.
        content:    Qualifier value (QID, string, or typed literal).

    Returns:
        Single-element list containing the qualifier dict.
    '''
    # Build Qualifer
    qualifier = [{"property":
                    {"id": identifier,
                     "data_type": data_type},
                     "value": 
                        {"type": "value",
                         "content": content}
                }]
    return qualifier

add_single_relation(statement, alt_statement=None, qualifier=None, reverse=False)

Add one statement per entry in subject[statement["relatant"]].

If the relatant item exists in the payload dictionary, uses statement['relation'] as the property; otherwise falls back to alt_statement with a string value.

Parameters:

Name	Type	Description	Default
statement		Dict with relation (property ID) and relatant (subject key) fields.	required
alt_statement		Fallback dict used when the relatant is not in the payload (optional).	None
qualifier		Pre-built qualifier list (optional).	None
reverse		If True, the relatant becomes the subject and the current item becomes the object.	False

Source code in MaRDMO/payload.py

def add_single_relation(
    self,
    statement,
    alt_statement = None,
    qualifier = None,
    reverse = False
):
    '''Add one statement per entry in ``subject[statement["relatant"]]``.

    If the relatant item exists in the payload dictionary, uses
    ``statement['relation']`` as the property; otherwise falls back to
    *alt_statement* with a string value.

    Args:
        statement:     Dict with ``relation`` (property ID) and
                       ``relatant`` (subject key) fields.
        alt_statement: Fallback dict used when the relatant is not in
                       the payload (optional).
        qualifier:     Pre-built qualifier list (optional).
        reverse:       If ``True``, the relatant becomes the subject and
                       the current item becomes the object.
    '''
    # Empty Qualifiers if none provided
    if qualifier is None:
        qualifier = []
    for entry in self.state.subject.get(statement['relatant'], {}).values():
        # Get Item Key
        entry_item = self.get_item_key(entry, 'object')
        if entry_item in self.state.dictionary:
            # Assign Object and Subject
            if reverse:
                subject_item, object_item = entry_item, self.state.subject_item
            else:
                subject_item, object_item = self.state.subject_item, entry_item
            # Add to Payload
            self.add_answer(
                verb = statement['relation'],
                object_and_type = [
                    object_item,
                    'wikibase-item',
                ],
                qualifier = qualifier,
                subject = subject_item
            )
        else:
            # Add to Payload
            self.add_answer(
                verb = alt_statement['relation'],
                object_and_type = [
                    entry.get(alt_statement['relatant']),
                    'string',
                ],
                qualifier = qualifier
            )

build_relation_check_query()

Build a SPARQL SELECT query that checks whether all RELATION entries already exist.

Returns:

Type	Description
	SPARQL query string that selects one boolean variable per relation
	(?RELATION0, ?RELATION1, …).

Source code in MaRDMO/payload.py

def build_relation_check_query(self):
    '''Build a SPARQL SELECT query that checks whether all RELATION entries already exist.

    Returns:
        SPARQL query string that selects one boolean variable per relation
        (``?RELATION0``, ``?RELATION1``, …).
    '''
    relation_keys = [k for k in self.state.dictionary if k.startswith('RELATION')]
    optional_blocks, bind_blocks = [], []

    for idx, key in enumerate(relation_keys):
        entry = self.state.dictionary[key]
        optional_block, bind_block = self._build_relation_block(idx, entry)
        if optional_block is None:
            continue
        optional_blocks.append(optional_block)
        bind_blocks.append(bind_block)

    query_body = '\n'.join(optional_blocks + bind_blocks)
    selectors = " ".join(f"?RELATION{idx}" for idx in range(len(relation_keys)))
    return f'\nSELECT {selectors} WHERE {{\n{query_body}\n}}'

check_math_relations_via_api(api_url)

Check existence of math-datatype RELATION statements via wbgetentities.

SPARQL returns math values as MathML, but portal stores LaTeX. This method queries the Wikibase API directly to compare raw LaTeX values.

Returns {relation_key: 'true'/'false'}.

Source code in MaRDMO/payload.py

def check_math_relations_via_api(self, api_url):
    '''Check existence of math-datatype RELATION statements via wbgetentities.

    SPARQL returns math values as MathML, but portal stores LaTeX.  This
    method queries the Wikibase API directly to compare raw LaTeX values.

    Returns {relation_key: 'true'/'false'}.
    '''
    relation_keys = [k for k in self.state.dictionary if k.startswith('RELATION')]
    math_entries  = {}  # key → (item_qid, prop_id, latex, qualifiers)

    for key in relation_keys:
        entry     = self.state.dictionary[key]
        statement = entry['payload']['statement']
        if statement['property']['data_type'] != 'math':
            continue
        item_key = entry['url'].split('/')[-2]
        item_qid = self.state.dictionary.get(item_key, {}).get('id', '')
        if not item_qid:
            continue  # new item → statement cannot exist yet
        math_entries[key] = (
            item_qid,
            statement['property']['id'],
            statement['value']['content'],
            statement.get('qualifiers', []),
        )

    if not math_entries:
        return {}

    # Fetch claims for all relevant items
    items_needed   = set(v[0] for v in math_entries.values())
    claims_by_item = {}
    chunk_size     = 50

    for i in range(0, len(items_needed), chunk_size):
        chunk = list(items_needed)[i:i + chunk_size]
        try:
            resp = requests.get(
                api_url,
                params={
                    'action': 'wbgetentities',
                    'ids':    '|'.join(chunk),
                    'props':  'claims',
                    'format': 'json',
                },
                headers={'User-Agent': 'MaRDMO (https://zib.de; reidelbach@zib.de)'},
                timeout=10,
            )
            resp.raise_for_status()
            for qid, entity in resp.json().get('entities', {}).items():
                claims_by_item[qid] = entity.get('claims', {})
        except requests.exceptions.RequestException as exc:
            _logger.warning("Math relation API check failed: %s", exc)

    result = {}
    for key, (item_qid, prop_id, latex, qualifiers) in math_entries.items():
        exists = 'false'
        for claim in claims_by_item.get(item_qid, {}).get(prop_id, []):
            claim_val = claim.get('mainsnak', {}).get('datavalue', {}).get('value', '')
            if claim_val != latex:
                continue
            if not qualifiers:
                exists = 'true'
                break
            # Check that all qualifiers match
            qual_match = True
            for q in qualifiers:
                q_prop = q['property']['id']
                q_val  = q['value']['content']
                # Resolve temporary item key to actual QID if available
                if q_val in self.state.dictionary:
                    q_val = self.state.dictionary[q_val].get('id', q_val)
                api_qual_vals = {
                    c.get('datavalue', {}).get('value', {}).get('id')
                    for c in claim.get('qualifiers', {}).get(q_prop, [])
                }
                if q_val not in api_qual_vals:
                    qual_match = False
                    break
            if qual_match:
                exists = 'true'
                break
        result[key] = exists

    return result

get_dictionary()

Return the complete payload dictionary (items, relations, and aliases).

The returned dict maps "Item<n>", "RELATION<n>", and "ALIAS<n>" keys to their respective payload entries. This is the top-level structure posted to the Wikibase REST API by :class:~MaRDMO.oauth2.OauthProviderMixin.

Source code in MaRDMO/payload.py

def get_dictionary(self):
    '''Return the complete payload dictionary (items, relations, and aliases).

    The returned dict maps ``"Item<n>"``, ``"RELATION<n>"``, and
    ``"ALIAS<n>"`` keys to their respective payload entries.  This is the
    top-level structure posted to the Wikibase REST API by
    :class:`~MaRDMO.oauth2.OauthProviderMixin`.
    '''
    # Get Target Dictionary
    target_dictionary = self.state.dictionary
    return target_dictionary

get_item_key(value, role='subject')

Look up the "Item<n>" key for value and optionally set it as the current subject.

Parameters:

Name	Type	Description	Default
value		Item dict with ID, Name, and Description fields.	required
role		'subject' (default) stores the item as the active subject for subsequent :meth:add_answer calls; 'object' just returns the key without updating state.	'subject'

Returns:

Type	Description
	The "Item<n>" key string for value.

Raises:

Type	Description
ValueError	If value is empty or missing Name/Description.

Source code in MaRDMO/payload.py

def get_item_key(self, value, role='subject'):
    """Look up the ``"Item<n>"`` key for *value* and optionally set it as the current subject.

    Args:
        value: Item dict with ``ID``, ``Name``, and ``Description`` fields.
        role:  ``'subject'`` (default) stores the item as the active
               subject for subsequent :meth:`add_answer` calls;
               ``'object'`` just returns the key without updating state.

    Returns:
        The ``"Item<n>"`` key string for *value*.

    Raises:
        ValueError: If *value* is empty or missing ``Name``/``Description``.
    """
    if not value:
        raise ValueError("Missing Item in Statement!")
    if not value.get('Name') or not value.get('Description'):
        raise ValueError("All Items need to have a 'Name' and 'Description'!")

    item_key = self._find_key_by_values(
        value['ID'],
        value['Name'],
        value['Description'],
    )

    if role == 'subject':
        self.state.subject = value
        self.state.subject_item = item_key

    return item_key

process_items()

Populate the payload dictionary with an entry for every item in user_items.

Dispatches each item to a source-specific handler based on the prefix of its ID field (mardi, wikidata, not found, no author found, no journal found). Existing MaRDI items are registered with their real QID; new items get an empty id and a seed list of statements. Raises :exc:ValueError for Wikidata or user-created items whose label/description combination already exists on the MaRDI Portal.

Source code in MaRDMO/payload.py

def process_items(self):
    """Populate the payload dictionary with an entry for every item in ``user_items``.

    Dispatches each item to a source-specific handler based on the prefix
    of its ``ID`` field (``mardi``, ``wikidata``, ``not found``,
    ``no author found``, ``no journal found``).  Existing MaRDI items are
    registered with their real QID; new items get an empty id and a
    seed list of statements.  Raises :exc:`ValueError` for Wikidata or
    user-created items whose label/description combination already exists
    on the MaRDI Portal.
    """
    handlers = {
        'mardi': lambda key, value: 
            self._add_entry(
                key,
                self._build_item(
                    value['ID'].split(':')[1],
                    value['Name'],
                    value['Description'],
                )
            ),
        'wikidata': lambda key, value: (
            self._check_mardi_and_raise(
                value['Name'],
                value['Description']
            ),
            self._add_entry(
                key,
                self._build_item(
                    '',
                    value['Name'],
                    value['Description'],
                    self._statement_by_id_type(
                        value,
                        'wikidata'))
            )
        ),
        'not found': lambda key, value: (
            self._check_mardi_and_raise(
                value['Name'],
                value['Description']
            ),
            self._add_entry(
                key,
                self._build_item(
                    '',
                    value['Name'],
                    value['Description'],
                    self._statement_by_id_type(
                        value,
                        'not found'
                    )
                )
            )
        ),
        'no author found': lambda key, value: (
            self._check_mardi_and_raise(
                value['Name'],
                value['Description']
            ),
            self._add_entry(
                key,
                self._build_item(
                    '',
                    value['Name'],
                    value['Description'],
                    self._statement_by_id_type(
                        value,
                        'no author found'
                    )
                )
            )
        ),
        'no journal found': lambda key, value: (
            self._check_mardi_and_raise(
                value['Name'],
                value['Description']
            ),
            self._add_entry(
                key,
                self._build_item(
                    '',
                    value['Name'],
                    value['Description'],
                    self._statement_by_id_type(
                        value,
                        'no journal found'
                    )
                )
            )
        ),
    }

    for key, value in self.user_items.items():
        if not value.get('ID'):
            continue
        for id_type, handler in handlers.items():
            if id_type in value['ID']:
                if id_type == 'no author found':
                    if value['zbmath'] or value['orcid']:
                        handler(key, value)
                else:
                    handler(key, value)
                break

set_class(name)

Tag the current subject item with an entity-class label.

Used by _export_* worker methods so that :func:~MaRDMO.helpers.compare_items can include the class in its output for grouping on the success page.

Parameters:

Name	Type	Description	Default
name	str	Human-readable class label (e.g. 'Mathematical Model').	required

Source code in MaRDMO/payload.py

def set_class(self, name: str) -> None:
    '''Tag the current subject item with an entity-class label.

    Used by ``_export_*`` worker methods so that :func:`~MaRDMO.helpers.compare_items`
    can include the class in its output for grouping on the success page.

    Args:
        name: Human-readable class label (e.g. ``'Mathematical Model'``).
    '''
    if self.state.subject_item:
        self.state.dictionary[self.state.subject_item]['class'] = name

PayloadState dataclass

Data Class to store the current state of the Payload.

Source code in MaRDMO/payload.py

@dataclass
class PayloadState:
    '''Data Class to store the current state of the Payload.'''
    counter: int = 0
    dictionary: dict = field(default_factory=dict)
    subject: dict = field(default_factory=dict)
    subject_item: Optional[str] = None

Store

Cache-backed progress and result store for MaRDMO background workers.

Background workers (preview, export) run asynchronously and need a way to communicate state and results back to the requesting view. This module provides a thin Django-cache wrapper that stores per-task progress percentages, status messages, and final payloads keyed by a UUID task identifier.

Provides:

• store_progress — write a progress percentage and status message
• retrieve_progress — read the current progress entry for a task
• store_result — persist the final worker output
• retrieve_result — fetch the final worker output

ProgressStore

Dict-like wrapper around Django's cache for job progress data.

This allows us to keep existing _progress_store[...] usages while backing the store with Django's cache backend.

Source code in MaRDMO/store.py

class ProgressStore:
    """Dict-like wrapper around Django's cache for job progress data.

    This allows us to keep existing `_progress_store[...]` usages while
    backing the store with Django's cache backend.
    """

    def __getitem__(self, job_id):
        '''Return the progress dict for *job_id*; raise :exc:`KeyError` if absent.'''
        value = get_progress_data(job_id)
        if value is None:
            raise KeyError(job_id)
        return value

    def __setitem__(self, job_id, value):
        '''Store *value* as the progress dict for *job_id* in the cache.'''
        set_progress_data(job_id, value)

    def get(self, job_id, default=None):
        '''Return the progress dict for *job_id*, or *default* if absent.'''
        return get_progress_data(job_id, default)

__getitem__(job_id)

Return the progress dict for job_id; raise :exc:KeyError if absent.

Source code in MaRDMO/store.py

def __getitem__(self, job_id):
    '''Return the progress dict for *job_id*; raise :exc:`KeyError` if absent.'''
    value = get_progress_data(job_id)
    if value is None:
        raise KeyError(job_id)
    return value

__setitem__(job_id, value)

Store value as the progress dict for job_id in the cache.

Source code in MaRDMO/store.py

def __setitem__(self, job_id, value):
    '''Store *value* as the progress dict for *job_id* in the cache.'''
    set_progress_data(job_id, value)

get(job_id, default=None)

Return the progress dict for job_id, or default if absent.

Source code in MaRDMO/store.py

def get(self, job_id, default=None):
    '''Return the progress dict for *job_id*, or *default* if absent.'''
    return get_progress_data(job_id, default)

clear_progress(job_id)

Delete the progress cache entry for job_id.

Source code in MaRDMO/store.py

def clear_progress(job_id):
    '''Delete the progress cache entry for *job_id*.'''
    cache.delete(_progress_cache_key(job_id))

get_progress_data(job_id, default=None)

Return the progress dict for job_id from the cache, or default if absent.

Source code in MaRDMO/store.py

def get_progress_data(job_id, default=None):
    '''Return the progress dict for *job_id* from the cache, or *default* if absent.'''
    return cache.get(_progress_cache_key(job_id), default)

set_progress_data(job_id, value, timeout=60 * 60)

Store value as the progress dict for job_id in the cache.

Parameters:

Name	Type	Description	Default
job_id		Unique job identifier string.	required
value		Progress dict to store.	required
timeout		Cache TTL in seconds (default 3600).	60 * 60

Source code in MaRDMO/store.py

def set_progress_data(job_id, value, timeout=60 * 60):
    '''Store *value* as the progress dict for *job_id* in the cache.

    Args:
        job_id:  Unique job identifier string.
        value:   Progress dict to store.
        timeout: Cache TTL in seconds (default 3600).
    '''
    cache.set(_progress_cache_key(job_id), value, timeout=timeout)

Adders

Module conaining functions writing retrieved metadata into an RDMO questionnaire.

After a background worker has collected entity data from external knowledge graphs, the results must be stored as RDMO Value objects so that the questionnaire reflects the fetched information. This module provides helpers that create or update those values for a given project and set of answers.

Provides:

• add_basics — write basic (label/description) values for a single entity entry
• add_entities — write a list of entities into a set of questionnaire answers
• add_new_entities — write newly user-created entities into the questionnaire
• add_relations_static — write relation values using a fixed property-to-question mapping
• add_relations_flexible — write relation values using a dynamic property-to-question mapping
• add_properties — write data-property values for an entity at a given URI
• add_references — write external-reference values for an entity

add_basics(project, text, questions, item_type, index=(None, None))

Parse the ID-question text and write label/description into the questionnaire.

Splits text (format "Label (Description) [source]") and stores the label and description in the Name and Description answer fields of the entity page identified by item_type.

Parameters:

Name	Type	Description	Default
project		RDMO project instance.	required
text		Human-readable ID string "Label (Description) [source]".	required
questions		Questions dict for the relevant catalog.	required
item_type		Entity type key in questions (e.g. "Research Field").	required
index		(set_index, set_prefix) tuple for the target page.	(None, None)

Returns:

Type	Description
	Tuple (label, description, source) extracted from text.

Source code in MaRDMO/adders.py

def add_basics(project, text, questions, item_type, index = (None, None)):
    '''Parse the ID-question text and write label/description into the questionnaire.

    Splits *text* (format ``"Label (Description) [source]"``) and stores the
    label and description in the Name and Description answer fields of the
    entity page identified by *item_type*.

    Args:
        project:   RDMO project instance.
        text:      Human-readable ID string ``"Label (Description) [source]"``.
        questions: Questions dict for the relevant catalog.
        item_type: Entity type key in *questions* (e.g. ``"Research Field"``).
        index:     ``(set_index, set_prefix)`` tuple for the target page.

    Returns:
        Tuple ``(label, description, source)`` extracted from *text*.
    '''

    # Extract Label, Description, Source from ID Question
    label, description, source = extract_parts(text)

    # Add Label to Questionnaire
    value_editor(
        project = project,
        uri = f'{BASE_URI}{questions[item_type]["Name"]["uri"]}',
        info = {
            'text': label,
            'set_index': index[0],
            'set_prefix': index[1]
        }
    )

    # Add Description to Questionnaire
    value_editor(
        project = project,
        uri = f'{BASE_URI}{questions[item_type]["Description"]["uri"]}',
        info = {
            'text': description,
            'set_index': index[0],
            'set_prefix': index[1]
        }
    )

    return label, description, source

add_entities(project, question_set, datas, source, prefix)

Ensure each item in datas has a page in question_set, creating one if absent.

Checks the existing questionnaire values (by external ID and by label/description) before adding a new set entry. Skips items that are already present under any of those checks.

Parameters:

Name	Type	Description	Default
project		RDMO project instance.	required
question_set		Attribute URI of the set question (e.g. the section root).	required
datas		Iterable of :class:~MaRDMO.models.Relatant instances.	required
source		Source tag written into the ID field (e.g. "mardi").	required
prefix		Label prefix for the set page (e.g. "AD").	required

Source code in MaRDMO/adders.py

def add_entities(project, question_set, datas, source, prefix):
    '''Ensure each item in *datas* has a page in *question_set*, creating one if absent.

    Checks the existing questionnaire values (by external ID and by
    label/description) before adding a new set entry.  Skips items that are
    already present under any of those checks.

    Args:
        project:      RDMO project instance.
        question_set: Attribute URI of the set question (e.g. the section root).
        datas:        Iterable of :class:`~MaRDMO.models.Relatant` instances.
        source:       Source tag written into the ID field (e.g. ``"mardi"``).
        prefix:       Label prefix for the set page (e.g. ``"AD"``).
    '''

    # Generate ID, Name and Description URL from Set URL
    question = {'id': f'{question_set}/id',
                'name': f'{question_set}/name',
                'description': f'{question_set}/description'}

    # Get existing Set and Item Information
    info = {'set_ids': get_id(project, question_set, ['set_index']),
            'value_ids': get_id(project, question['id'], ['external_id']),
            'texts': get_id(project, question['id'], ['text']),
            'names': get_id(project, question['name'], ['text']),
            'descs': get_id(project, question['description'], ['text'])}

    # Add Item to Questionnaire
    idx = max(info['set_ids'], default = -1) + 1

    for data in datas:
        # Label Description String
        name_desc = f'{data.label} ({data.description})'
        # Check if Item already in Questionnaire via ID Question
        check_id = any(
            name_desc in text
            for text in info['texts']
            )
        # Check if Item already in Questionnaire via Name/Description Question
        check_name_desc = any(
            name_desc in f'{name} ({desc})'
            for name, desc in zip(info['names'], info['descs'])
            )
        # If Item not already in Questionnaire
        if data.id not in info['value_ids'] and not check_id and not check_name_desc:
            # Set up Page in Questionnaire
            value_editor(
                project = project,
                uri = question_set,
                info = {
                    'text': f"{prefix}{int(idx)+1}",
                    'set_index': idx
                }
            )
            # Add ID Values
            value_editor(
                project = project,
                uri = question['id'],
                info = {
                    'text': f'{data.label} ({data.description}) [{source}]',
                    'external_id': f"{data.id}",
                    'set_index': idx
                }
            )

            # Update Index and existing Items
            idx += 1
            info['value_ids'].append(data.id)

add_new_entities(project, question_set, datas, prefix)

Ensure each user-defined item in datas has a page in question_set.

Like :func:add_entities, but for user-created items (no external ID). Deduplicates by label/description only and marks new entries with external_id = "not found".

Parameters:

Name	Type	Description	Default
project		RDMO project instance.	required
question_set		Attribute URI of the set question.	required
datas		Iterable of :class:~MaRDMO.models.Relatant instances.	required
prefix		Label prefix for the set page.	required

Source code in MaRDMO/adders.py

def add_new_entities(project, question_set, datas, prefix):
    '''Ensure each user-defined item in *datas* has a page in *question_set*.

    Like :func:`add_entities`, but for user-created items (no external ID).
    Deduplicates by label/description only and marks new entries with
    ``external_id = "not found"``.

    Args:
        project:      RDMO project instance.
        question_set: Attribute URI of the set question.
        datas:        Iterable of :class:`~MaRDMO.models.Relatant` instances.
        prefix:       Label prefix for the set page.
    '''

    # Generate ID, Name and Description URL from Set URL
    question = {'id': f'{question_set}/id',
                'name': f'{question_set}/name',
                'description': f'{question_set}/description'}

    # Get existing Set and Item Information
    info = {'set_ids': get_id(project, question_set, ['set_index']),
            'names': get_id(project, question['name'], ['text']),
            'descs': get_id(project, question['description'], ['text'])}

    # Add Publication to Questionnaire
    idx = max(info['set_ids'], default = -1) + 1
    for data in datas:
        # Label Description String
        name_desc = f'{data.label} ({data.description})'
        # Check if Item already in Questionnaire via Name/Description Question
        check_name_desc = any(
            name_desc == f'{name} ({desc})'
            for name, desc in zip(info['names'], info['descs'])
            )
        # If Item not already in Questionnaire
        if not check_name_desc:
            # Set up Page
            value_editor(
                project = project,
                uri = question_set,
                info = {
                    'text': f"{prefix}{int(idx)+1}",
                    'set_index': idx
                }
            )
            # Add ID Values
            value_editor(
                project = project,
                uri = question['id'],
                info = {
                    'text': 'not found',
                    'external_id': 'not found',
                    'set_index': idx
                }
            )
            # Add Name Values
            value_editor(
                project = project,
                uri = question['name'],
                info = {
                    'text': data.label,
                    'set_prefix': idx
                }
            )
            # Add Description Values
            value_editor(
                project = project,
                uri = question['description'],
                info = {
                    'text': data.description,
                    'set_prefix': idx
                }
            )

            # Update Index
            idx += 1

add_properties(project, data, uri, set_prefix)

Write the data-property option values from data.properties into the questionnaire.

Iterates over the {collection_index: [option_uri, …]} mapping in data.properties and calls :func:~MaRDMO.helpers.value_editor for each entry at set_index=0.

Parameters:

Name	Type	Description	Default
project		RDMO project instance.	required
data		Dataclass instance with a properties attribute.	required
uri		Attribute URI of the data-property collection question.	required
set_prefix		Set-prefix of the parent entity page.	required

Source code in MaRDMO/adders.py

def add_properties(project, data, uri, set_prefix):
    '''Write the data-property option values from *data.properties* into the questionnaire.

    Iterates over the ``{collection_index: [option_uri, …]}`` mapping in
    ``data.properties`` and calls :func:`~MaRDMO.helpers.value_editor` for
    each entry at ``set_index=0``.

    Args:
        project:    RDMO project instance.
        data:       Dataclass instance with a ``properties`` attribute.
        uri:        Attribute URI of the data-property collection question.
        set_prefix: Set-prefix of the parent entity page.
    '''

    for key, value in data.properties.items():
        value_editor(
            project = project,
            uri  = uri,
            info = {
                'option': Option.objects.get(uri=value[0]),
                'collection_index': key,
                'set_index': 0,
                'set_prefix': set_prefix
            }
        )

add_references(project, data, uri, set_index=0, set_prefix=None)

Write the reference entries from data.reference into the questionnaire.

Does nothing when data.reference is empty. Each entry in the {collection_index: [option_uri, text]} mapping is stored as an option value at the given set_index / set_prefix.

Parameters:

Name	Type	Description	Default
project		RDMO project instance.	required
data		Dataclass instance with a reference attribute.	required
uri		Attribute URI of the reference collection question.	required
set_index		Set-index of the parent entity page (default 0).	0
set_prefix		Set-prefix of the parent entity page (optional).	None

Source code in MaRDMO/adders.py

def add_references(project, data, uri, set_index = 0, set_prefix = None):
    '''Write the reference entries from *data.reference* into the questionnaire.

    Does nothing when ``data.reference`` is empty.  Each entry in the
    ``{collection_index: [option_uri, text]}`` mapping is stored as an option
    value at the given *set_index* / *set_prefix*.

    Args:
        project:    RDMO project instance.
        data:       Dataclass instance with a ``reference`` attribute.
        uri:        Attribute URI of the reference collection question.
        set_index:  Set-index of the parent entity page (default ``0``).
        set_prefix: Set-prefix of the parent entity page (optional).
    '''
    if not data.reference:
        return

    for key, value in data.reference.items():
        value_editor(
            project = project,
            uri  = uri,
            info = {
                'text': value[1],
                'option': Option.objects.get(uri=value[0]),
                'collection_index': key,
                'set_index': set_index,
                'set_prefix': set_prefix
            }
        )

add_relations_flexible(project, data, props, index, statement)

Write flexible (typed) relations from data into the questionnaire.

For each relatant in data.<prop> (where prop ∈ props['keys']), checks whether the (relation-type, relatant) pair already exists. If not, writes the relation-type option and the relatant text/ID, handling optional order-number and assumption qualifiers.

Parameters:

Name	Type	Description	Default
project		RDMO project instance.	required
data		Dataclass instance whose attributes hold lists of relatants.	required
props		Dict with keys 'keys' (attribute names) and 'mapping' (:class:~MaRDMO.helpers.PropertyRegistry mapping prop name → relation URL).	required
index		Dict containing 'set_prefix'; updated with 'set_prefix_reduced' and 'idx' in place.	required
statement		Dict with keys 'relation', 'relatant', and optionally 'order' and 'assumption' (attribute URIs).	required

Source code in MaRDMO/adders.py

def add_relations_flexible(project, data, props, index, statement):
    '''Write flexible (typed) relations from *data* into the questionnaire.

    For each relatant in ``data.<prop>`` (where *prop* ∈ ``props['keys']``),
    checks whether the (relation-type, relatant) pair already exists.  If not,
    writes the relation-type option and the relatant text/ID, handling optional
    order-number and assumption qualifiers.

    Args:
        project:   RDMO project instance.
        data:      Dataclass instance whose attributes hold lists of relatants.
        props:     Dict with keys ``'keys'`` (attribute names) and ``'mapping'``
                   (:class:`~MaRDMO.helpers.PropertyRegistry` mapping prop
                   name → relation URL).
        index:     Dict containing ``'set_prefix'``; updated with
                   ``'set_prefix_reduced'`` and ``'idx'`` in place.
        statement: Dict with keys ``'relation'``, ``'relatant'``, and
                   optionally ``'order'`` and ``'assumption'`` (attribute URIs).
    '''

    # Get existing Set, Item and Relation Information
    info = {'set_prefix_ids': get_id(project, statement['relatant'], ['set_prefix']),
            'set_index_ids': get_id(project, statement['relatant'], ['set_index']),
            'collection_ids': get_id(project, statement['relatant'], ['collection_index']),
            'value_ids': get_id(project, statement['relatant'], ['external_id']),
            'texts': get_id(project, statement['relatant'], ['text']),
            'rels': get_id(project, statement['relation'], ['option_uri'])}

    # Get reduced set prefix ids
    index.update({'set_prefix_reduced': reduce_prefix(index['set_prefix'])})

    # Get relevant set index ids
    ids = relevant_set_ids(info, index['set_prefix_reduced'])

    # Set initial value of counter
    index.update({'idx': initialize_counter(ids)})

    # Add Relations and Relatants
    for prop in props['keys']:
        inner_idx = 0
        assumption_store = {}
        order_number_store = {}
        for value in getattr(data, prop):
            assumption_index = None
            order_number_index = None
            # Get Source and Label Description String
            source, _ = value.id.split(':')

            # Check if Relation / Relatant Combination exists (flexible relation)
            matches = relation_exists(
                value = value,
                set_prefix_red = index['set_prefix_reduced'],
                info = info,
                relation_id = props['mapping'].get(key=prop)["url"]
            )

            if matches:
                # Continue if existing
                continue

            # Add Order Number
            if statement.get('order') and hasattr(value, 'other') and value.other:
                if value.other not in order_number_store:
                    index['idx'] +=1
                    inner_idx = 0
                    order_number_store.update({value.other: index['idx']})
                order_number_index = order_number_store.get(value.other)
                # Add Order Number to Questionnaire
                value_editor(
                    project = project,
                    uri = statement['order'],
                    info = {
                        'text': value.other,
                        'set_index': order_number_index,
                        'set_prefix': index['set_prefix']
                    }
                )

            # Add Assumption
            if statement.get('assumption') and hasattr(value, 'qualifier') and value.qualifier:
                if value.qualifier not in assumption_store:
                    index['idx'] +=1
                    inner_idx = 0
                    assumption_store.update({value.qualifier: index['idx']})
                assumption_index = assumption_store.get(value.qualifier)
                # Get Assumptions
                assumption_dict = process_qualifier(value.qualifier)
                # Add Assumptions
                for assumption_key, assumption_value in assumption_dict.items():
                    value_editor(
                        project = project,
                        uri = statement['assumption'],
                        info = {
                            'text': "{label} ({description}) [{source}]".format_map(
                                assumption_value
                            ),
                            'external_id': assumption_value['id'],
                            'collection_index': assumption_key,
                            'set_index': assumption_index,
                            'set_prefix': index['set_prefix']
                        }
                    )

            # Add Relation to Questionnaire
            value_editor(
                project = project,
                uri = statement['relation'],
                info = {
                    'option': Option.objects.get(uri=props['mapping'].get(key=prop)["url"]),
                    'collection_index': None,
                    'set_index': order_number_index or assumption_index or index['idx'],
                    'set_prefix': index['set_prefix']
                }
            )

            # Add Relatant to Questionnaire
            value_editor(
                project = project,
                uri = statement['relatant'],
                info = {
                    'text': f"{value.label} ({value.description}) [{source}]",
                    'external_id': value.id,
                    'collection_index': inner_idx,
                    'set_index': order_number_index or assumption_index or index['idx'],
                    'set_prefix': index['set_prefix']
                }
            )

            # Update existing IDs, Texts, and Relations
            info['value_ids'].append(value.id)
            info['set_prefix_ids'].append(index['set_prefix_reduced'])
            info['texts'].append(f"{value.label} ({value.description}) [{source}]")
            info['rels'].append(props['mapping'].get(key=prop)["url"])

            inner_idx += 1

        # Update index
        index['idx'] += 1

add_relations_static(project, data, props, index, statement)

Write static (fixed-type) relations from data into the questionnaire.

Iterates over the relatant lists named by props['keys'] and, for each relatant not already present (checked by external ID and text), adds a new collection entry to statement['relatant'].

When statement contains a 'platform' key and a relatant carries a qualifier attribute, platform values are written first and the relatant is grouped under the corresponding platform set-index with inner_idx as its collection-index. The grouping key is (qualifier, other) so that two relatants with the same platform but different other payloads land in separate sets. When statement also contains a 'parameter' key and the relatant has a non-empty other field, each ' || '-split segment of other is written as a separate parameter entry. Without a qualifier the original behaviour is preserved: set_index=0, collection_index=index['idx'].

Parameters:

Name	Type	Description	Default
project		RDMO project instance.	required
data		Dataclass instance whose attributes hold lists of relatants.	required
props		Dict with key 'keys' listing attribute names on data.	required
index		Dict containing 'set_prefix'; 'set_prefix_reduced' and 'idx' are computed and written back into this dict.	required
statement		Dict with key 'relatant' (attribute URI of the collection question) and optional keys 'platform' (attribute URI of the platform question) and 'parameter' (attribute URI of the parameter question).	required

Source code in MaRDMO/adders.py

def add_relations_static(project, data, props, index, statement):
    '''Write static (fixed-type) relations from *data* into the questionnaire.

    Iterates over the relatant lists named by ``props['keys']`` and, for each
    relatant not already present (checked by external ID and text), adds a
    new collection entry to ``statement['relatant']``.

    When ``statement`` contains a ``'platform'`` key and a relatant carries a
    ``qualifier`` attribute, platform values are written first and the relatant
    is grouped under the corresponding platform set-index with ``inner_idx``
    as its collection-index.  The grouping key is ``(qualifier, other)`` so
    that two relatants with the same platform but different ``other`` payloads
    land in separate sets.  When ``statement`` also contains a ``'parameter'``
    key and the relatant has a non-empty ``other`` field, each ``' || '``-split
    segment of ``other`` is written as a separate parameter entry.
    Without a qualifier the original behaviour is preserved:
    ``set_index=0``, ``collection_index=index['idx']``.

    Args:
        project:   RDMO project instance.
        data:      Dataclass instance whose attributes hold lists of relatants.
        props:     Dict with key ``'keys'`` listing attribute names on *data*.
        index:     Dict containing ``'set_prefix'``; ``'set_prefix_reduced'``
                   and ``'idx'`` are computed and written back into this dict.
        statement: Dict with key ``'relatant'`` (attribute URI of the
                   collection question) and optional keys ``'platform'``
                   (attribute URI of the platform question) and
                   ``'parameter'`` (attribute URI of the parameter question).
    '''

    # Get existing Set and Item Information
    info = {'set_prefix_ids': get_id(project, statement['relatant'], ['set_prefix']),
            'set_index_ids': get_id(project, statement['relatant'], ['set_index']),
            'collection_ids': get_id(project, statement['relatant'], ['collection_index']),
            'value_ids': get_id(project, statement['relatant'], ['external_id']),
            'texts': get_id(project, statement['relatant'], ['text'])}

    # Get reduced set_prefixes
    index.update({'set_prefix_reduced': reduce_prefix(index['set_prefix'])})

    # Set initial value of counter
    index.update({'idx': initialize_counter(info['collection_ids'])})

    # Add Relations and Relatants
    for prop in props['keys']:
        inner_idx = 0
        for value in getattr(data, prop):
            assumption_index = None

            # Get Source and Label Description String
            source, _ = value.id.split(':')

            # Check if Relatant exists
            matches = relation_exists(
                value = value,
                set_prefix_red = index['set_prefix_reduced'],
                info = info)

            if matches:
                # Continue if existing
                continue

            # Add Assumption
            if statement.get('platform') and isinstance(value, ProcessStepUsage):
                index['idx'] += 1
                inner_idx = 0
                assumption_index = index['idx']

                if value.qualifier:
                    qual_source, _ = value.qualifier.split(':')
                    value_editor(
                        project = project,
                        uri = statement['platform'],
                        info = {
                            'text': f"{value.qualifier_label} ({value.qualifier_description}) [{qual_source}]",
                            'external_id': value.qualifier,
                            'collection_index': 0,
                            'set_index': assumption_index,
                            'set_prefix': index['set_prefix']
                        }
                    )

                if statement.get('hardware') and value.hardware:
                    hw_source, _ = value.hardware.split(':')
                    value_editor(
                        project = project,
                        uri = statement['hardware'],
                        info = {
                            'text': f"{value.hardware_label} ({value.hardware_description}) [{hw_source}]",
                            'external_id': value.hardware,
                            'collection_index': 0,
                            'set_index': assumption_index,
                            'set_prefix': index['set_prefix']
                        }
                    )

                if statement.get('documentation'):
                    doc_idx = 0
                    for doc_val in (value.doi, value.url):
                        if doc_val:
                            value_editor(
                                project = project,
                                uri = statement['documentation'],
                                info = {
                                    'text': doc_val[1],
                                    'option': Option.objects.get(uri=doc_val[0]),
                                    'collection_index': doc_idx,
                                    'set_index': assumption_index,
                                    'set_prefix': index['set_prefix']
                                }
                            )
                            doc_idx += 1

                if statement.get('parameter') and value.parameters:
                    for i, param in enumerate(value.parameters.split(' || ')):
                        value_editor(
                            project = project,
                            uri = statement['parameter'],
                            info = {
                                'text': param,
                                'collection_index': i,
                                'set_index': assumption_index,
                                'set_prefix': index['set_prefix']
                            }
                        )

            # Add Relatant to Questionnaire
            value_editor(
                project = project,
                uri = statement['relatant'],
                info = {
                    'text': f"{value.label} ({value.description}) [{source}]",
                    'external_id': value.id,
                    'collection_index': inner_idx if assumption_index is not None else index['idx'],
                    'set_index': assumption_index or 0,
                    'set_prefix': index['set_prefix']
                }
            )

            # Update Index
            if assumption_index is not None:
                inner_idx += 1
            else:
                index['idx'] += 1

            # Update existing IDs, Texts, and Relations
            info['value_ids'].append(value.id)
            info['set_prefix_ids'].append(index['set_prefix_reduced'])
            info['texts'].append(f"{value.label} ({value.description}) [{source}]")

Builders

Factory function that builds the attribute-URI-to-handler dispatch map.

The router (router.py) needs to know which handler method to call for each RDMO attribute URI that arrives in a signal. This module assembles that map once at startup by resolving question URIs from the RDMO database and pairing them with the appropriate Information handler methods from each sub-package.

Provides:

• build_post_save_handler_set — post-save dispatch dict (all catalogs)
• build_post_delete_handler_set — post-delete dispatch dict (workflow + publication catalogs)

build_post_delete_handler_set()

Build and return the post-delete attribute-URI-to-handler dispatch set.

Covers:

• Publication catalog: when a Publication value set is deleted, the dependent citation values are cleaned up via :meth:~MaRDMO.publication.handlers.Information.publication_delete.

Returns:

Name	Type	Description
dict		Nested mapping of the form
		{catalog_slug: {absolute_attribute_uri: handler_method}},
		structured identically to the map returned by
		func:build_post_save_handler_set.

Source code in MaRDMO/builders.py

def build_post_delete_handler_set():
    """Build and return the post-delete attribute-URI-to-handler dispatch set.

    Covers:

    - Publication catalog: when a Publication value set is deleted, the dependent
      citation values are cleaned up via
      :meth:`~MaRDMO.publication.handlers.Information.publication_delete`.

    Returns:
        dict: Nested mapping of the form
        ``{catalog_slug: {absolute_attribute_uri: handler_method}}``,
        structured identically to the map returned by
        :func:`build_post_save_handler_set`.
    """

    base = BASE_URI
    questions_publication = get_questions('publication')
    publication = PublicationInformation()

    pub_set_uri = f"{base}{questions_publication['Publication']['uri']}"

    return {
        'mardmo-model-catalog': {
            pub_set_uri: publication.publication_delete,
        },
        'mardmo-model-basics-catalog': {
            pub_set_uri: publication.publication_delete,
        },
        'mardmo-algorithm-catalog': {
            pub_set_uri: publication.publication_delete,
        },
        'mardmo-interdisciplinary-workflow-catalog': {
            pub_set_uri: publication.publication_delete,
        },
    }

build_post_save_handler_set()

Build and return the post-save attribute-URI-to-handler dispatch set.

Loads question URI configurations from the RDMO database for all four catalogs (model, algorithm, workflow, publication), instantiates the corresponding Information handler objects, and maps each relevant attribute URI to the correct handler method.

Returns:

Name	Type	Description
dict		Nested mapping of the form
		{catalog_slug: {absolute_attribute_uri: handler_method}}.
		Each inner dict is passed to the router so that incoming
		value_created / value_updated signals can be dispatched
		without any per-signal lookups.

Source code in MaRDMO/builders.py

def build_post_save_handler_set():
    """Build and return the post-save attribute-URI-to-handler dispatch set.

    Loads question URI configurations from the RDMO database for all four
    catalogs (model, algorithm, workflow, publication), instantiates the
    corresponding ``Information`` handler objects, and maps each relevant
    attribute URI to the correct handler method.

    Returns:
        dict: Nested mapping of the form
        ``{catalog_slug: {absolute_attribute_uri: handler_method}}``.
        Each inner dict is passed to the router so that incoming
        ``value_created`` / ``value_updated`` signals can be dispatched
        without any per-signal lookups.
    """

    base = BASE_URI
    handler_map = {}

    # Questions
    questions_model = get_questions('model')
    questions_algorithm = get_questions('algorithm')
    questions_workflow = get_questions('workflow')
    questions_publication = get_questions('publication')

    # Information Classes
    model = ModelInformation()
    algorithm = AlgorithmInformation()
    workflow = WorkflowInformation()
    publication = PublicationInformation()
    general = GeneralInformation()

    # Model handlers
    handler_map.update({
        'mardmo-model-catalog': {
            f"{base}{questions_model['Research Field']['ID']['uri']}":
                model.field,
            f"{base}{questions_model['Research Problem']['ID']['uri']}":
                model.problem,
            f"{base}{questions_model['Quantity']['ID']['uri']}":
                model.quantity,
            f"{base}{questions_model['Mathematical Formulation']['ID']['uri']}":
                model.formulation,
            f"{base}{questions_model['Task']['ID']['uri']}":
                model.task,
            f"{base}{questions_model['Mathematical Model']['ID']['uri']}":
                model.model,
            f'{base}{questions_model["Task"]["QRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Task"]["MFRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Mathematical Formulation"]["Element Quantity"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Quantity"]["Element Quantity"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Mathematical Model"]["MFRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Mathematical Model"]["Assumption"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Task"]["Assumption"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Mathematical Formulation"]["Assumption"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Mathematical Formulation"]["MFRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Research Problem"]["RFRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Mathematical Model"]["RPRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Mathematical Model"]["TRelatant"]["uri"]}':
                general.relation,
            f"{base}{questions_publication['Publication']['ID']['uri']}":
                publication.citation,
        }
    })

    # Model handlers
    handler_map.update({
        'mardmo-model-basics-catalog': {
            f"{base}{questions_model['Research Problem']['ID']['uri']}":
                model.problem,
            f"{base}{questions_model['Task']['ID']['uri']}":
                model.task,
            f"{base}{questions_model['Mathematical Model']['ID']['uri']}":
                model.model,
            f"{base}{questions_model['Mathematical Formulation']['ID']['uri']}":
                model.formulation,
            f'{base}{questions_model["Mathematical Model"]["RPRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Mathematical Model"]["TRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Mathematical Model"]["MFRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Mathematical Model"]["Assumption"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Task"]["MFRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Task"]["Assumption"]["uri"]}':
                general.relation,
            f'{base}{questions_model["Mathematical Formulation"]["Assumption"]["uri"]}':
                general.relation,
            f"{base}{questions_publication['Publication']['ID']['uri']}":
                publication.citation,
        }
    })

    # Algorithm handlers
    handler_map.update({
        'mardmo-algorithm-catalog': {
            f"{base}{questions_algorithm['Benchmark']['ID']['uri']}":
                algorithm.benchmark,
            f"{base}{questions_algorithm['Software']['ID']['uri']}":
                algorithm.software,
            f"{base}{questions_algorithm['Problem']['ID']['uri']}":
                algorithm.problem,
            f"{base}{questions_algorithm['Algorithm']['ID']['uri']}":
                algorithm.algorithm,
            f'{base}{questions_algorithm["Problem"]["BRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_algorithm["Software"]["BRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_algorithm["Algorithm"]["PRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_algorithm["Algorithm"]["SRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_algorithm["Software"]["Dependency"]["uri"]}':
                general.relation,
            f"{base}{questions_publication['Publication']['ID']['uri']}":
                publication.citation,
        }
    })

    # Workflow handlers
    handler_map.update({
        'mardmo-interdisciplinary-workflow-catalog': {
            f"{base}{questions_workflow['Workflow']['ID']['uri']}":
                workflow.workflow,
            f"{base}{questions_workflow['Algorithm']['ID']['uri']}":
                workflow.algorithm,
            f"{base}{questions_workflow['Workflow']['Model']['uri']}":
                workflow.model,
            f"{base}{questions_workflow['Software']['ID']['uri']}":
                workflow.software,
            f"{base}{questions_workflow['Hardware']['ID']['uri']}":
                workflow.hardware,
            f"{base}{questions_workflow['Hardware']['CPU']['uri']}":
                workflow.processor_cores,
            f"{base}{questions_workflow['Data Set']['ID']['uri']}":
                workflow.data_set,
            f"{base}{questions_workflow['Process Step']['ID']['uri']}":
                workflow.process_step,
            f'{base}{questions_workflow["Process Step"]["Algorithm"]["uri"]}':
                general.relation,
            f'{base}{questions_workflow["Process Step"]["Hardware"]["uri"]}':
                general.relation,
            f'{base}{questions_workflow["Process Step"]["Input"]["uri"]}':
                general.relation,
            f'{base}{questions_workflow["Process Step"]["Output"]["uri"]}':
                general.relation,
            f'{base}{questions_workflow["Workflow"]["PSRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_workflow["Process Step"]["Software"]["uri"]}':
                general.relation,
            f'{base}{questions_workflow["Algorithm"]["SRelatant"]["uri"]}':
                general.relation,
            f'{base}{questions_workflow["Software"]["Dependency"]["uri"]}':
                general.relation,
            f"{base}{questions_publication['Publication']['ID']['uri']}":
                publication.citation,
        }
    })

    return handler_map