Technical docs

Views#

Django views.

PeerReviewView#

View for the reviewer role of the Open Peer Review process.

Bases: LoginRequiredMixin, View

A view handling the peer review of metadata. This view supports loading, parsing, sorting metadata, and handling GET and POST requests for peer review.

Source code in dataedit/views.py

class TablePeerReviewView(LoginRequiredMixin, View):
    """
    A view handling the peer review of metadata. This view supports loading,
    parsing, sorting metadata, and handling GET and POST requests for peer review.
    """

    def load_json(self, schema: str, table: str, review_id=None):
        """
        Load JSON metadata from the database. If the review_id is available
        then load the metadata form the peer review instance and not from the
        table. This avoids changes to the metadata that is or was reviewed.

        Args:
            schema (str): The schema of the table.
            table (str): The name of the table.
            review_id (int): Id of a peer review in the django database

        Returns:
            dict: Loaded oemetadata.
        """
        metadata = {}
        if review_id is None:
            metadata = load_metadata_from_db(table=table)
        elif review_id:
            opr = PeerReviewManager.get_opr_by_id(opr_id=review_id)
            metadata = opr.oemetadata

        return metadata

    def load_json_schema(self):
        """
        Load the JSON schema used for validating metadata.

        Note:
            Update this method if a new oemetadata version is released.

        Returns:
            dict: JSON schema.
        """
        json_schema = OEMETADATA_V20_SCHEMA
        return json_schema

    def parse_keys(self, val, old=""):
        """
        Recursively parse keys from a nested dictionary or list and return them
        as a list of dictionaries.

        Args:
            val (dict or list): The input dictionary or list to parse.
            old (str, optional): The prefix for nested keys. Defaults to an
                empty string.

        Returns:
            list: A list of dictionaries, each containing 'field' and 'value'
                keys.
        """
        lines = []
        if isinstance(val, dict):
            for k in val.keys():
                lines += self.parse_keys(val[k], old + "." + str(k))
        elif isinstance(val, list):
            if not val:
                # handles empty list
                lines += [{"field": old[1:], "value": str(val)}]
            else:
                for i, k in enumerate(val):
                    lines += self.parse_keys(
                        k, old + "." + str(i)
                    )  # handles user value
        else:
            lines += [{"field": old[1:], "value": str(val)}]
        return lines

    def sort_in_category(self, schema: str, table: str, oemetadata):
        """
        Group flattened OEMetadata v2 fields into thematic buckets and attach
        placeholders required by the review UI.

        Each entry has six keys:
        {
          "field": "<dot-path>",
          "label": "<display label without 'resources.<idx>.'>",
          "value": "<current value>",
          "newValue": "",
          "reviewer_suggestion": "",
          "suggestion_comment": ""
        }
        """

        flattened = self.parse_keys(oemetadata)
        flattened = [
            item for item in flattened if item["field"].startswith("resources.")
        ]

        bucket_map = {
            "spatial": "spatial",
            "temporal": "temporal",
            "sources": "source",
            "licenses": "license",
        }

        def make_label(dot_path: str) -> str:
            # remove leading resources.<idx>.
            trimmed = re.sub(r"^resources\.[0-9]+\.", "", dot_path)
            parts = trimmed.split(".")
            out = []
            for p in parts:
                if p in {"@id", "@type"}:
                    out.append(p)
                else:
                    out.append(p.replace("_", " "))
            if out:
                out[0] = out[0][:1].upper() + out[0][1:]
            return " ".join(out)

        tmp = defaultdict(list)

        for item in flattened:
            raw_key = item["field"]
            parts = raw_key.split(".")

            if parts[0] == "resources" and len(parts) >= 3:
                root = parts[2]
            else:
                root = parts[0]

            bucket = bucket_map.get(root, "general")

            tmp[bucket].append(
                {
                    "field": raw_key,
                    "label": make_label(raw_key),
                    "value": item["value"],
                    "newValue": "",
                    "reviewer_suggestion": "",
                    "suggestion_comment": "",
                }
            )

        return {
            "general": tmp["general"],
            "spatial": tmp["spatial"],
            "temporal": tmp["temporal"],
            "source": tmp["source"],
            "license": tmp["license"],
        }

    def get_all_field_descriptions(self, json_schema, prefix=""):
        """
        Collects the field title, descriptions, examples, and badge information
        for each field of the oemetadata from the JSON schema and prepares them
        for further processing.

        Args:
            json_schema (dict): The JSON schema to extract field descriptions
                from.
            prefix (str, optional): The prefix for nested keys. Defaults to an
                empty string.

        Returns:
            dict: A dictionary containing field descriptions, examples, and
                other information.
        """

        field_descriptions = {}

        def extract_descriptions(properties, prefix=""):
            for field, value in properties.items():
                key = f"{prefix}.{field}" if prefix else field

                if any(
                    attr in value
                    for attr in ["description", "examples", "example", "badge", "title"]
                ):
                    field_descriptions[key] = {}
                    if "description" in value:
                        field_descriptions[key]["description"] = value["description"]
                    # Prefer v2 "examples" (array) over v1 "example" (single value)
                    if "examples" in value and value["examples"]:
                        # v2: first item of the examples array
                        field_descriptions[key]["example"] = value["examples"][0]
                    elif "example" in value:
                        # v1 fallback
                        field_descriptions[key]["example"] = value["example"]
                    if "badge" in value:
                        field_descriptions[key]["badge"] = value["badge"]
                    if "title" in value:
                        field_descriptions[key]["title"] = value["title"]
                if "properties" in value:
                    new_prefix = f"{prefix}.{field}" if prefix else field
                    extract_descriptions(value["properties"], new_prefix)
                if "items" in value:
                    new_prefix = f"{prefix}.{field}" if prefix else field
                    if "properties" in value["items"]:
                        extract_descriptions(value["items"]["properties"], new_prefix)

        extract_descriptions(json_schema["properties"], prefix)
        return field_descriptions

    def get(
        self,
        request: HttpRequest,
        schema: str,
        table: str,
        review_id: int | None = None,
    ) -> HttpResponse:
        """
        Handle GET requests for peer review.
        Loads necessary data and renders the review template.

        Args:
            request (HttpRequest): The incoming HTTP GET request.
            schema (str): The schema of the table.
            table (str): The name of the table.
            review_id (int, optional): The ID of the review. Defaults to None.

        Returns:
            HttpResponse: Rendered HTML response.
        """
        # review_state = PeerReview.is_finished  # TODO: Use later
        json_schema = self.load_json_schema()
        can_add = False
        table_obj = Table.load(name=table)
        field_descriptions = self.get_all_field_descriptions(json_schema)

        # Check user permissions
        user: login_models.myuser = request.user  # type: ignore
        if not user.is_anonymous:
            level = user.get_table_permission_level(table_obj)
            can_add = level >= login_models.WRITE_PERM

        oemetadata = self.load_json(schema, table, review_id)
        metadata = self.sort_in_category(
            schema, table, oemetadata=oemetadata
        )  # Generate URL for peer_review_reviewer
        if review_id is not None:
            url_peer_review = reverse(
                "dataedit:peer_review_reviewer",
                kwargs={"schema": schema, "table": table, "review_id": review_id},
            )
            opr_review = PeerReviewManager.get_opr_by_id(opr_id=review_id)
            existing_review = opr_review.review.get("reviews", [])
            review_finished = opr_review.is_finished
            categories = [
                "general",
                "spatial",
                "temporal",
                "source",
                "license",
            ]
            state_dict = process_review_data(
                review_data=existing_review, metadata=metadata, categories=categories
            )
        else:
            url_peer_review = reverse(
                "dataedit:peer_review_create", kwargs={"schema": schema, "table": table}
            )
            # existing_review={}
            state_dict = None
            review_finished = None

        config_data = {
            "can_add": can_add,
            "url_peer_review": url_peer_review,
            "url_table": reverse(
                "dataedit:view", kwargs={"schema": schema, "table": table}
            ),
            "topic": schema,
            "table": table,
            "review_finished": review_finished,
            "review_id": review_id,
        }
        context_meta = {
            # need this here as json.dumps breaks the template syntax access
            # like {{ config.table }} now you can use {{ table }}
            "table": table,
            "topic": schema,
            "config": json.dumps(config_data),
            "meta": metadata,
            "json_schema": json_schema,
            "field_descriptions_json": json.dumps(field_descriptions),
            "state_dict": json.dumps(state_dict),
            "review_finished": review_finished,
            "review_id": review_id,
        }
        return render(request, "dataedit/opr_review.html", context=context_meta)

    def post(
        self, request: HttpRequest, schema: str, table: str, review_id=None
    ) -> HttpResponse:
        """
        Handle POST requests for submitting reviews by the reviewer.

        This method:
        - Creates (or saves) reviews in the PeerReview table.
        - Updates the review finished attribute in the dataedit.Tables table,
            indicating that the table can be moved from the model draft topic.

        Missing parts:
        - once the opr is finished (all field reviews agreed on)
        - merge field review results to metadata on table
        - awarde a badge
            - is field filled in?
            - calculate the badge by comparing filled fields
              and the badges form metadata schema

        Args:
            request (HttpRequest): The incoming HTTP POST request.
            schema (str): The schema of the table.
            table (str): The name of the table.
            review_id (int, optional): The ID of the review. Defaults to None.

        Returns:
            HttpResponse: Rendered HTML response for the review.

        Raises:
            JsonResponse: If any error occurs, a JsonResponse containing the
            error message is raised.

        Note:
            - There are some missing parts in this method. Once the review process
                is finished (all field reviews agreed on), it should merge field
                review results to metadata on the table and award a badge based
                on certain criteria.
            - A notification should be sent to the user if he/she can't review tables
            for which he/she is the table holder (TODO).
            - After a review is finished, the table's metadata is updated, and the table
            can be moved to a different schema or topic (TODO).
        """
        context = {}
        user: login_models.myuser = request.user  # type: ignore

        # get the review data and additional application metadata
        # from user peer review submit/save
        review_data = json.loads(request.body)
        if review_id:
            contributor_review = PeerReview.objects.filter(id=review_id).first()
            if contributor_review:
                contributor_review_data = contributor_review.review.get("reviews", [])
                review_data["reviewData"]["reviews"].extend(contributor_review_data)

        # The type can be "save" or "submit" as this triggers different behavior
        review_post_type = review_data.get("reviewType")
        # The opr datamodel that includes the field review data and metadata
        review_datamodel = review_data.get("reviewData")
        review_finished = review_datamodel.get("reviewFinished")
        # TODO: Send a notification to the user that he can't review tables
        # he is the table holder.
        if review_post_type == "delete":
            return delete_peer_review(review_id)

        contributor = PeerReviewManager.load_contributor(table=table)

        if contributor is not None:
            # Überprüfen, ob ein aktiver PeerReview existiert
            active_peer_review = PeerReview.load(table=table)
            if active_peer_review is None or active_peer_review.is_finished:
                # Kein aktiver PeerReview vorhanden
                # oder der aktive PeerReview ist abgeschlossen
                table_review = PeerReview(
                    table=table,
                    is_finished=review_finished,
                    review=review_datamodel,
                    reviewer=user,
                    contributor=contributor,
                    oemetadata=load_metadata_from_db(table=table),
                )
                table_review.save(review_type=review_post_type)
            else:
                # Aktiver PeerReview ist vorhanden ... aktualisieren
                current_review_data = active_peer_review.review
                merged_review_data = merge_field_reviews(
                    current_json=current_review_data, new_json=review_datamodel
                )

                # Set new review values and update existing review
                active_peer_review.review = merged_review_data
                active_peer_review.reviewer = user  # type: ignore
                active_peer_review.contributor = contributor
                active_peer_review.update(review_type=review_post_type)
        else:
            error_msg = (
                "Failed to retrieve any user that identifies "
                f"as table holder for the current table: {table}!"
            )
            return JsonResponse({"error": error_msg}, status=400)

        # TODO: Check for schema/topic as reviewed finished also indicates the table
        # needs to be or has to be moved.
        if review_finished is True:
            review_table = Table.load(name=table)
            review_table.set_is_reviewed()
            metadata = self.load_json(schema, table, review_id=review_id)
            updated_metadata = recursive_update(metadata, review_data)
            save_metadata_to_db(schema, table, updated_metadata)
            active_peer_review = PeerReview.load(table=table)

            if active_peer_review:
                updated_oemetadata = recursive_update(
                    active_peer_review.oemetadata, review_data
                )
                active_peer_review.oemetadata = updated_oemetadata
                active_peer_review.save()

            # TODO: also update reviewFinished in review datamodel json
            # logging.INFO(f"Table {table.name} is now reviewed and can be moved
            # to the destination schema.")

        return render(request, "dataedit/opr_review.html", context=context)

`get(request, schema, table, review_id=None)` #

Handle GET requests for peer review. Loads necessary data and renders the review template.

Parameters:

Name	Type	Description	Default
`request`	`HttpRequest`	The incoming HTTP GET request.	required
`schema`	`str`	The schema of the table.	required
`table`	`str`	The name of the table.	required
`review_id`	`int`	The ID of the review. Defaults to None.	`None`

Returns:

Name	Type	Description
`HttpResponse`	`HttpResponse`	Rendered HTML response.

Source code in dataedit/views.py

def get(
    self,
    request: HttpRequest,
    schema: str,
    table: str,
    review_id: int | None = None,
) -> HttpResponse:
    """
    Handle GET requests for peer review.
    Loads necessary data and renders the review template.

    Args:
        request (HttpRequest): The incoming HTTP GET request.
        schema (str): The schema of the table.
        table (str): The name of the table.
        review_id (int, optional): The ID of the review. Defaults to None.

    Returns:
        HttpResponse: Rendered HTML response.
    """
    # review_state = PeerReview.is_finished  # TODO: Use later
    json_schema = self.load_json_schema()
    can_add = False
    table_obj = Table.load(name=table)
    field_descriptions = self.get_all_field_descriptions(json_schema)

    # Check user permissions
    user: login_models.myuser = request.user  # type: ignore
    if not user.is_anonymous:
        level = user.get_table_permission_level(table_obj)
        can_add = level >= login_models.WRITE_PERM

    oemetadata = self.load_json(schema, table, review_id)
    metadata = self.sort_in_category(
        schema, table, oemetadata=oemetadata
    )  # Generate URL for peer_review_reviewer
    if review_id is not None:
        url_peer_review = reverse(
            "dataedit:peer_review_reviewer",
            kwargs={"schema": schema, "table": table, "review_id": review_id},
        )
        opr_review = PeerReviewManager.get_opr_by_id(opr_id=review_id)
        existing_review = opr_review.review.get("reviews", [])
        review_finished = opr_review.is_finished
        categories = [
            "general",
            "spatial",
            "temporal",
            "source",
            "license",
        ]
        state_dict = process_review_data(
            review_data=existing_review, metadata=metadata, categories=categories
        )
    else:
        url_peer_review = reverse(
            "dataedit:peer_review_create", kwargs={"schema": schema, "table": table}
        )
        # existing_review={}
        state_dict = None
        review_finished = None

    config_data = {
        "can_add": can_add,
        "url_peer_review": url_peer_review,
        "url_table": reverse(
            "dataedit:view", kwargs={"schema": schema, "table": table}
        ),
        "topic": schema,
        "table": table,
        "review_finished": review_finished,
        "review_id": review_id,
    }
    context_meta = {
        # need this here as json.dumps breaks the template syntax access
        # like {{ config.table }} now you can use {{ table }}
        "table": table,
        "topic": schema,
        "config": json.dumps(config_data),
        "meta": metadata,
        "json_schema": json_schema,
        "field_descriptions_json": json.dumps(field_descriptions),
        "state_dict": json.dumps(state_dict),
        "review_finished": review_finished,
        "review_id": review_id,
    }
    return render(request, "dataedit/opr_review.html", context=context_meta)

`get_all_field_descriptions(json_schema, prefix='')` #

Collects the field title, descriptions, examples, and badge information for each field of the oemetadata from the JSON schema and prepares them for further processing.

Parameters:

Name	Type	Description	Default
`json_schema`	`dict`	The JSON schema to extract field descriptions from.	required
`prefix`	`str`	The prefix for nested keys. Defaults to an empty string.	`''`

Returns:

Name	Type	Description
`dict`		A dictionary containing field descriptions, examples, and other information.

Source code in dataedit/views.py

def get_all_field_descriptions(self, json_schema, prefix=""):
    """
    Collects the field title, descriptions, examples, and badge information
    for each field of the oemetadata from the JSON schema and prepares them
    for further processing.

    Args:
        json_schema (dict): The JSON schema to extract field descriptions
            from.
        prefix (str, optional): The prefix for nested keys. Defaults to an
            empty string.

    Returns:
        dict: A dictionary containing field descriptions, examples, and
            other information.
    """

    field_descriptions = {}

    def extract_descriptions(properties, prefix=""):
        for field, value in properties.items():
            key = f"{prefix}.{field}" if prefix else field

            if any(
                attr in value
                for attr in ["description", "examples", "example", "badge", "title"]
            ):
                field_descriptions[key] = {}
                if "description" in value:
                    field_descriptions[key]["description"] = value["description"]
                # Prefer v2 "examples" (array) over v1 "example" (single value)
                if "examples" in value and value["examples"]:
                    # v2: first item of the examples array
                    field_descriptions[key]["example"] = value["examples"][0]
                elif "example" in value:
                    # v1 fallback
                    field_descriptions[key]["example"] = value["example"]
                if "badge" in value:
                    field_descriptions[key]["badge"] = value["badge"]
                if "title" in value:
                    field_descriptions[key]["title"] = value["title"]
            if "properties" in value:
                new_prefix = f"{prefix}.{field}" if prefix else field
                extract_descriptions(value["properties"], new_prefix)
            if "items" in value:
                new_prefix = f"{prefix}.{field}" if prefix else field
                if "properties" in value["items"]:
                    extract_descriptions(value["items"]["properties"], new_prefix)

    extract_descriptions(json_schema["properties"], prefix)
    return field_descriptions

`load_json(schema, table, review_id=None)` #

Load JSON metadata from the database. If the review_id is available then load the metadata form the peer review instance and not from the table. This avoids changes to the metadata that is or was reviewed.

Parameters:

Name	Type	Description	Default
`schema`	`str`	The schema of the table.	required
`table`	`str`	The name of the table.	required
`review_id`	`int`	Id of a peer review in the django database	`None`

Returns:

Name	Type	Description
`dict`		Loaded oemetadata.

Source code in dataedit/views.py

def load_json(self, schema: str, table: str, review_id=None):
    """
    Load JSON metadata from the database. If the review_id is available
    then load the metadata form the peer review instance and not from the
    table. This avoids changes to the metadata that is or was reviewed.

    Args:
        schema (str): The schema of the table.
        table (str): The name of the table.
        review_id (int): Id of a peer review in the django database

    Returns:
        dict: Loaded oemetadata.
    """
    metadata = {}
    if review_id is None:
        metadata = load_metadata_from_db(table=table)
    elif review_id:
        opr = PeerReviewManager.get_opr_by_id(opr_id=review_id)
        metadata = opr.oemetadata

    return metadata

`load_json_schema()` #

Load the JSON schema used for validating metadata.

Note

Update this method if a new oemetadata version is released.

Returns:

Name	Type	Description
`dict`		JSON schema.

Source code in dataedit/views.py

def load_json_schema(self):
    """
    Load the JSON schema used for validating metadata.

    Note:
        Update this method if a new oemetadata version is released.

    Returns:
        dict: JSON schema.
    """
    json_schema = OEMETADATA_V20_SCHEMA
    return json_schema

`parse_keys(val, old='')` #

Recursively parse keys from a nested dictionary or list and return them as a list of dictionaries.

Parameters:

Name	Type	Description	Default
`val`	`dict or list`	The input dictionary or list to parse.	required
`old`	`str`	The prefix for nested keys. Defaults to an empty string.	`''`

Returns:

Name	Type	Description
`list`		A list of dictionaries, each containing 'field' and 'value' keys.

Source code in dataedit/views.py

def parse_keys(self, val, old=""):
    """
    Recursively parse keys from a nested dictionary or list and return them
    as a list of dictionaries.

    Args:
        val (dict or list): The input dictionary or list to parse.
        old (str, optional): The prefix for nested keys. Defaults to an
            empty string.

    Returns:
        list: A list of dictionaries, each containing 'field' and 'value'
            keys.
    """
    lines = []
    if isinstance(val, dict):
        for k in val.keys():
            lines += self.parse_keys(val[k], old + "." + str(k))
    elif isinstance(val, list):
        if not val:
            # handles empty list
            lines += [{"field": old[1:], "value": str(val)}]
        else:
            for i, k in enumerate(val):
                lines += self.parse_keys(
                    k, old + "." + str(i)
                )  # handles user value
    else:
        lines += [{"field": old[1:], "value": str(val)}]
    return lines

`post(request, schema, table, review_id=None)` #

Handle POST requests for submitting reviews by the reviewer.

This method: - Creates (or saves) reviews in the PeerReview table. - Updates the review finished attribute in the dataedit.Tables table, indicating that the table can be moved from the model draft topic.

Missing parts: - once the opr is finished (all field reviews agreed on) - merge field review results to metadata on table - awarde a badge - is field filled in? - calculate the badge by comparing filled fields and the badges form metadata schema

Parameters:

Name	Type	Description	Default
`request`	`HttpRequest`	The incoming HTTP POST request.	required
`schema`	`str`	The schema of the table.	required
`table`	`str`	The name of the table.	required
`review_id`	`int`	The ID of the review. Defaults to None.	`None`

Returns:

Name	Type	Description
`HttpResponse`	`HttpResponse`	Rendered HTML response for the review.

Raises:

Type	Description
`JsonResponse`	If any error occurs, a JsonResponse containing the

Note

There are some missing parts in this method. Once the review process is finished (all field reviews agreed on), it should merge field review results to metadata on the table and award a badge based on certain criteria.
A notification should be sent to the user if he/she can't review tables for which he/she is the table holder (TODO).
After a review is finished, the table's metadata is updated, and the table can be moved to a different schema or topic (TODO).

Source code in dataedit/views.py

def post(
    self, request: HttpRequest, schema: str, table: str, review_id=None
) -> HttpResponse:
    """
    Handle POST requests for submitting reviews by the reviewer.

    This method:
    - Creates (or saves) reviews in the PeerReview table.
    - Updates the review finished attribute in the dataedit.Tables table,
        indicating that the table can be moved from the model draft topic.

    Missing parts:
    - once the opr is finished (all field reviews agreed on)
    - merge field review results to metadata on table
    - awarde a badge
        - is field filled in?
        - calculate the badge by comparing filled fields
          and the badges form metadata schema

    Args:
        request (HttpRequest): The incoming HTTP POST request.
        schema (str): The schema of the table.
        table (str): The name of the table.
        review_id (int, optional): The ID of the review. Defaults to None.

    Returns:
        HttpResponse: Rendered HTML response for the review.

    Raises:
        JsonResponse: If any error occurs, a JsonResponse containing the
        error message is raised.

    Note:
        - There are some missing parts in this method. Once the review process
            is finished (all field reviews agreed on), it should merge field
            review results to metadata on the table and award a badge based
            on certain criteria.
        - A notification should be sent to the user if he/she can't review tables
        for which he/she is the table holder (TODO).
        - After a review is finished, the table's metadata is updated, and the table
        can be moved to a different schema or topic (TODO).
    """
    context = {}
    user: login_models.myuser = request.user  # type: ignore

    # get the review data and additional application metadata
    # from user peer review submit/save
    review_data = json.loads(request.body)
    if review_id:
        contributor_review = PeerReview.objects.filter(id=review_id).first()
        if contributor_review:
            contributor_review_data = contributor_review.review.get("reviews", [])
            review_data["reviewData"]["reviews"].extend(contributor_review_data)

    # The type can be "save" or "submit" as this triggers different behavior
    review_post_type = review_data.get("reviewType")
    # The opr datamodel that includes the field review data and metadata
    review_datamodel = review_data.get("reviewData")
    review_finished = review_datamodel.get("reviewFinished")
    # TODO: Send a notification to the user that he can't review tables
    # he is the table holder.
    if review_post_type == "delete":
        return delete_peer_review(review_id)

    contributor = PeerReviewManager.load_contributor(table=table)

    if contributor is not None:
        # Überprüfen, ob ein aktiver PeerReview existiert
        active_peer_review = PeerReview.load(table=table)
        if active_peer_review is None or active_peer_review.is_finished:
            # Kein aktiver PeerReview vorhanden
            # oder der aktive PeerReview ist abgeschlossen
            table_review = PeerReview(
                table=table,
                is_finished=review_finished,
                review=review_datamodel,
                reviewer=user,
                contributor=contributor,
                oemetadata=load_metadata_from_db(table=table),
            )
            table_review.save(review_type=review_post_type)
        else:
            # Aktiver PeerReview ist vorhanden ... aktualisieren
            current_review_data = active_peer_review.review
            merged_review_data = merge_field_reviews(
                current_json=current_review_data, new_json=review_datamodel
            )

            # Set new review values and update existing review
            active_peer_review.review = merged_review_data
            active_peer_review.reviewer = user  # type: ignore
            active_peer_review.contributor = contributor
            active_peer_review.update(review_type=review_post_type)
    else:
        error_msg = (
            "Failed to retrieve any user that identifies "
            f"as table holder for the current table: {table}!"
        )
        return JsonResponse({"error": error_msg}, status=400)

    # TODO: Check for schema/topic as reviewed finished also indicates the table
    # needs to be or has to be moved.
    if review_finished is True:
        review_table = Table.load(name=table)
        review_table.set_is_reviewed()
        metadata = self.load_json(schema, table, review_id=review_id)
        updated_metadata = recursive_update(metadata, review_data)
        save_metadata_to_db(schema, table, updated_metadata)
        active_peer_review = PeerReview.load(table=table)

        if active_peer_review:
            updated_oemetadata = recursive_update(
                active_peer_review.oemetadata, review_data
            )
            active_peer_review.oemetadata = updated_oemetadata
            active_peer_review.save()

        # TODO: also update reviewFinished in review datamodel json
        # logging.INFO(f"Table {table.name} is now reviewed and can be moved
        # to the destination schema.")

    return render(request, "dataedit/opr_review.html", context=context)

`sort_in_category(schema, table, oemetadata)` #

Group flattened OEMetadata v2 fields into thematic buckets and attach placeholders required by the review UI.

Each entry has six keys: { "field": "", "label": ".'>", "value": "", "newValue": "", "reviewer_suggestion": "", "suggestion_comment": "" }

Source code in dataedit/views.py

def sort_in_category(self, schema: str, table: str, oemetadata):
    """
    Group flattened OEMetadata v2 fields into thematic buckets and attach
    placeholders required by the review UI.

    Each entry has six keys:
    {
      "field": "<dot-path>",
      "label": "<display label without 'resources.<idx>.'>",
      "value": "<current value>",
      "newValue": "",
      "reviewer_suggestion": "",
      "suggestion_comment": ""
    }
    """

    flattened = self.parse_keys(oemetadata)
    flattened = [
        item for item in flattened if item["field"].startswith("resources.")
    ]

    bucket_map = {
        "spatial": "spatial",
        "temporal": "temporal",
        "sources": "source",
        "licenses": "license",
    }

    def make_label(dot_path: str) -> str:
        # remove leading resources.<idx>.
        trimmed = re.sub(r"^resources\.[0-9]+\.", "", dot_path)
        parts = trimmed.split(".")
        out = []
        for p in parts:
            if p in {"@id", "@type"}:
                out.append(p)
            else:
                out.append(p.replace("_", " "))
        if out:
            out[0] = out[0][:1].upper() + out[0][1:]
        return " ".join(out)

    tmp = defaultdict(list)

    for item in flattened:
        raw_key = item["field"]
        parts = raw_key.split(".")

        if parts[0] == "resources" and len(parts) >= 3:
            root = parts[2]
        else:
            root = parts[0]

        bucket = bucket_map.get(root, "general")

        tmp[bucket].append(
            {
                "field": raw_key,
                "label": make_label(raw_key),
                "value": item["value"],
                "newValue": "",
                "reviewer_suggestion": "",
                "suggestion_comment": "",
            }
        )

    return {
        "general": tmp["general"],
        "spatial": tmp["spatial"],
        "temporal": tmp["temporal"],
        "source": tmp["source"],
        "license": tmp["license"],
    }

PeerRreviewContributorView#

View for the contributor role of the Open Peer Review process.

Bases: TablePeerReviewView

A view handling the contributor's side of the peer review process. This view supports rendering the review template and handling GET and POST requests for contributor's review.

Source code in dataedit/views.py

class TablePeerRreviewContributorView(TablePeerReviewView):
    """
    A view handling the contributor's side of the peer review process.
    This view supports rendering the review template and handling GET and
    POST requests for contributor's review.
    """

    def get(
        self, request: HttpRequest, schema: str, table: str, review_id: int
    ) -> HttpResponse:
        """
        Handle GET requests for contributor's review. Loads necessary data and
        renders the contributor review template.

        Args:
            request (HttpRequest): The incoming HTTP GET request.
            schema (str): The schema of the table.
            table (str): The name of the table.
            review_id (int): The ID of the review.

        Returns:
            HttpResponse: Rendered HTML response for contributor review.
        """
        can_add = False
        peer_review = PeerReview.objects.get(id=review_id)
        table_obj = Table.load(peer_review.table)
        user: login_models.myuser = request.user  # type: ignore
        if not user.is_anonymous:
            level = user.get_table_permission_level(table_obj)
            can_add = level >= login_models.WRITE_PERM
        oemetadata = self.load_json(schema, table, review_id)
        metadata = self.sort_in_category(schema, table, oemetadata=oemetadata)
        json_schema = self.load_json_schema()
        field_descriptions = self.get_all_field_descriptions(json_schema)
        review_data = (peer_review.review or {}).get("reviews", [])

        categories = [
            "general",
            "spatial",
            "temporal",
            "source",
            "license",
        ]
        state_dict = process_review_data(
            review_data=review_data, metadata=metadata, categories=categories
        )
        context_meta = {
            "config": json.dumps(
                {
                    "can_add": can_add,
                    "url_peer_review": reverse(
                        "dataedit:peer_review_contributor",
                        kwargs={
                            "schema": schema,
                            "table": table,
                            "review_id": review_id,
                        },
                    ),
                    "url_table": reverse(
                        "dataedit:view", kwargs={"schema": schema, "table": table}
                    ),
                    "topic": schema,
                    "table": table,
                }
            ),
            "table": table,
            "topic": schema,
            "meta": metadata,
            "json_schema": json_schema,
            "field_descriptions_json": json.dumps(field_descriptions),
            "state_dict": json.dumps(state_dict),
        }
        return render(request, "dataedit/opr_contributor.html", context=context_meta)

    def post(
        self, request: HttpRequest, schema: str, table: str, review_id: int
    ) -> HttpResponse:
        """
        Handle POST requests for contributor's review. Merges and updates
        the review data in the PeerReview table.

        Args:
            request (HttpRequest): The incoming HTTP POST request.
            schema (str): The schema of the table.
            table (str): The name of the table.
            review_id (int): The ID of the review.

        Returns:
            HttpResponse: Rendered HTML response for contributor review.

        """

        context = {}
        if request.method == "POST":
            review_data = json.loads(request.body)
            review_post_type = review_data.get("reviewType")
            review_datamodel = review_data.get("reviewData")
            current_opr = PeerReviewManager.get_opr_by_id(opr_id=review_id)
            existing_reviews = current_opr.review
            merged_review = merge_field_reviews(
                current_json=existing_reviews, new_json=review_datamodel
            )

            current_opr.review = merged_review
            current_opr.update(review_type=review_post_type)

        return render(request, "dataedit/opr_contributor.html", context=context)

`get(request, schema, table, review_id)` #

Handle GET requests for contributor's review. Loads necessary data and renders the contributor review template.

Parameters:

Name	Type	Description	Default
`request`	`HttpRequest`	The incoming HTTP GET request.	required
`schema`	`str`	The schema of the table.	required
`table`	`str`	The name of the table.	required
`review_id`	`int`	The ID of the review.	required

Returns:

Name	Type	Description
`HttpResponse`	`HttpResponse`	Rendered HTML response for contributor review.

Source code in dataedit/views.py

def get(
    self, request: HttpRequest, schema: str, table: str, review_id: int
) -> HttpResponse:
    """
    Handle GET requests for contributor's review. Loads necessary data and
    renders the contributor review template.

    Args:
        request (HttpRequest): The incoming HTTP GET request.
        schema (str): The schema of the table.
        table (str): The name of the table.
        review_id (int): The ID of the review.

    Returns:
        HttpResponse: Rendered HTML response for contributor review.
    """
    can_add = False
    peer_review = PeerReview.objects.get(id=review_id)
    table_obj = Table.load(peer_review.table)
    user: login_models.myuser = request.user  # type: ignore
    if not user.is_anonymous:
        level = user.get_table_permission_level(table_obj)
        can_add = level >= login_models.WRITE_PERM
    oemetadata = self.load_json(schema, table, review_id)
    metadata = self.sort_in_category(schema, table, oemetadata=oemetadata)
    json_schema = self.load_json_schema()
    field_descriptions = self.get_all_field_descriptions(json_schema)
    review_data = (peer_review.review or {}).get("reviews", [])

    categories = [
        "general",
        "spatial",
        "temporal",
        "source",
        "license",
    ]
    state_dict = process_review_data(
        review_data=review_data, metadata=metadata, categories=categories
    )
    context_meta = {
        "config": json.dumps(
            {
                "can_add": can_add,
                "url_peer_review": reverse(
                    "dataedit:peer_review_contributor",
                    kwargs={
                        "schema": schema,
                        "table": table,
                        "review_id": review_id,
                    },
                ),
                "url_table": reverse(
                    "dataedit:view", kwargs={"schema": schema, "table": table}
                ),
                "topic": schema,
                "table": table,
            }
        ),
        "table": table,
        "topic": schema,
        "meta": metadata,
        "json_schema": json_schema,
        "field_descriptions_json": json.dumps(field_descriptions),
        "state_dict": json.dumps(state_dict),
    }
    return render(request, "dataedit/opr_contributor.html", context=context_meta)

`post(request, schema, table, review_id)` #

Handle POST requests for contributor's review. Merges and updates the review data in the PeerReview table.

Parameters:

Name	Type	Description	Default
`request`	`HttpRequest`	The incoming HTTP POST request.	required
`schema`	`str`	The schema of the table.	required
`table`	`str`	The name of the table.	required
`review_id`	`int`	The ID of the review.	required

Returns:

Name	Type	Description
`HttpResponse`	`HttpResponse`	Rendered HTML response for contributor review.

Source code in dataedit/views.py

def post(
    self, request: HttpRequest, schema: str, table: str, review_id: int
) -> HttpResponse:
    """
    Handle POST requests for contributor's review. Merges and updates
    the review data in the PeerReview table.

    Args:
        request (HttpRequest): The incoming HTTP POST request.
        schema (str): The schema of the table.
        table (str): The name of the table.
        review_id (int): The ID of the review.

    Returns:
        HttpResponse: Rendered HTML response for contributor review.

    """

    context = {}
    if request.method == "POST":
        review_data = json.loads(request.body)
        review_post_type = review_data.get("reviewType")
        review_datamodel = review_data.get("reviewData")
        current_opr = PeerReviewManager.get_opr_by_id(opr_id=review_id)
        existing_reviews = current_opr.review
        merged_review = merge_field_reviews(
            current_json=existing_reviews, new_json=review_datamodel
        )

        current_opr.review = merged_review
        current_opr.update(review_type=review_post_type)

    return render(request, "dataedit/opr_contributor.html", context=context)

Helper Functions#

Separated functionality that can be imported in other modules. It contains several functions that help with recurring tasks in the peer review system.

Provide helper functionality for views to reduce code lines in views.py make the codebase more modular.

SPDX-FileCopyrightText: 2025 Christian Winger https://github.com/wingechr © Öko-Institut e.V. SPDX-FileCopyrightText: 2025 Daryna Barabanova https://github.com/Darynarli © Reiner Lemoine Institut SPDX-FileCopyrightText: 2025 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-FileCopyrightText: 2025 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-FileCopyrightText: 2025 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-FileCopyrightText: 2025 user https://github.com/Darynarli © Reiner Lemoine Institut SPDX-License-Identifier: AGPL-3.0-or-later

`add_tag(name, color)` #

Parameters:

Name	Type	Description	Default
`name`	`str`	max 40 character tag text	required
`color`	`str`	hexadecimal color code, eg #aaf0f0	required

Source code in dataedit/helper.py

def add_tag(name: str, color: str) -> None:
    """
    Args:
        name(str): max 40 character tag text
        color(str): hexadecimal color code, eg #aaf0f0
    """
    Tag(name=name, color=Tag.color_from_hex(color)).save()

`change_requests(schema, table)` #

Loads the dataedit admin interface :param request: :return:

Source code in dataedit/helper.py

def change_requests(schema, table):
    """
    Loads the dataedit admin interface
    :param request:
    :return:
    """
    # I want to display old and new data, if different.

    display_message = None
    api_columns = actions.get_column_changes(reviewed=False, schema=schema, table=table)
    api_constraints = actions.get_constraints_changes(
        reviewed=False, schema=schema, table=table
    )

    data = dict()

    data["api_columns"] = {}
    data["api_constraints"] = {}

    keyword_whitelist = [
        "column_name",
        "c_table",
        "c_schema",
        "reviewed",
        "changed",
        "id",
    ]

    old_description = actions.describe_columns(schema, table)

    for change in api_columns:
        name = change["column_name"]
        id = change["id"]

        # Identifing over 'new'.
        if change.get("new_name") is not None:
            change["column_name"] = change["new_name"]

        old_cd = old_description.get(name)

        data["api_columns"][id] = {}
        data["api_columns"][id]["old"] = {}

        if old_cd is not None:
            old = api.parser.parse_scolumnd_from_columnd(
                schema, table, name, old_description.get(name)
            )

            for key in list(change):
                value = change[key]
                if key not in keyword_whitelist and (
                    value is None or value == old[key]
                ):
                    old.pop(key)
                    change.pop(key)
            data["api_columns"][id]["old"] = old
        else:
            data["api_columns"][id]["old"]["c_schema"] = schema
            data["api_columns"][id]["old"]["c_table"] = table
            data["api_columns"][id]["old"]["column_name"] = name

        data["api_columns"][id]["new"] = change

    for i in range(len(api_constraints)):
        value = api_constraints[i]
        id = value.get("id")
        if (
            value.get("reference_table") is None
            or value.get("reference_column") is None
        ):
            value.pop("reference_table")
            value.pop("reference_column")

        data["api_constraints"][id] = value

    display_style = [
        "c_schema",
        "c_table",
        "column_name",
        "not_null",
        "data_type",
        "reference_table",
        "constraint_parameter",
        "reference_column",
        "action",
        "constraint_type",
        "constraint_name",
    ]

    return {
        "data": data,
        "display_items": display_style,
        "display_message": display_message,
    }

`delete_peer_review(review_id)` #

Remove Peer Review by review_id. Args: review_id (int): ID review.

Returns:

Name	Type	Description
`JsonResponse`		JSON response about successful deletion or error.

Source code in dataedit/helper.py

def delete_peer_review(review_id):
    """
    Remove Peer Review by review_id.
    Args:
        review_id (int): ID review.

    Returns:
        JsonResponse: JSON response about successful deletion or error.
    """
    if review_id:
        peer_review = PeerReview.objects.filter(id=review_id).first()
        if peer_review:
            peer_review.delete()
            return JsonResponse({"message": "PeerReview successfully deleted."})
        else:
            return JsonResponse({"error": "PeerReview not found."}, status=404)
    else:
        return JsonResponse({"error": "Review ID is required."}, status=400)

`edit_tag(id, name, color)` #

Parameters:

Name	Type	Description	Default
`id`	`int`	tag id	required
`name`	`str`	max 40 character tag text	required
`color`	`str`	hexadecimal color code, eg #aaf0f0	required

Raises: sqlalchemy.exc.IntegrityError if name is not ok

Source code in dataedit/helper.py

def edit_tag(id: str, name: str, color: str) -> None:
    """
    Args:
        id(int): tag id
        name(str): max 40 character tag text
        color(str): hexadecimal color code, eg #aaf0f0
    Raises:
        sqlalchemy.exc.IntegrityError if name is not ok

    """
    tag = Tag.objects.get(pk=id)
    tag.name = name
    tag.color = Tag.color_from_hex(color)
    tag.save()

`find_tables(topic_name=None, query_string=None, tag_ids=None)` #

find tables given search criteria

Parameters:

Name	Type	Description	Default
`topic_name`	`str`	only tables in this topic	`None`
`query_string`	`str`	user search term	`None`
`tag_ids`	`list`	list of tag ids	`None`

Returns:

Type	Description
`QuerySet[Table]`	QuerySet of Table objetcs

Source code in dataedit/helper.py

def find_tables(
    topic_name: str | None = None,
    query_string: str | None = None,
    tag_ids: list[str] | None = None,
) -> QuerySet[Table]:
    """find tables given search criteria

    Args:
        topic_name (str, optional): only tables in this topic
        query_string (str, optional): user search term
        tag_ids (list, optional): list of tag ids

    Returns:
        QuerySet of Table objetcs
    """

    tables = Table.objects

    tables = tables.filter(is_sandbox=False)

    if topic_name:
        # TODO: WINGECHR: model_draft is not a topic, but currently,
        # frontend still usses it to filter / search for unpublished data
        if topic_name == TODO_PSEUDO_TOPIC_DRAFT:
            tables = tables.filter(is_publish=False)
        else:
            tables = tables.filter(topics__pk=topic_name)

    if query_string:  # filter by search terms
        tables = tables.filter(
            Q(
                search=SearchQuery(
                    " & ".join(p + ":*" for p in re.findall(r"[\w]+", query_string)),
                    search_type="raw",
                )
            )
        )

    if tag_ids:  # filter by tags:
        # find tables (in schema), that use all of the tags
        # by adding a filter for each tag
        # (instead of all at once, which would be OR)
        for tag_id in tag_ids:
            tables = tables.filter(tags__pk=tag_id)

    return tables

`get_all_tags(schema_name=None, table_name=None)` #

Load all tags of a specific table :param schema: Name of a schema :param table: Name of a table :return:

Source code in dataedit/helper.py

def get_all_tags(
    schema_name: str | None = None, table_name: str | None = None
) -> QuerySet[Tag]:
    """
    Load all tags of a specific table
    :param schema: Name of a schema
    :param table: Name of a table
    :return:
    """
    if table_name:
        tags = Table.objects.get(name=table_name).tags.all()
    else:
        tags = Tag.objects.all()

    return tags

`get_column_description(schema, table)` #

Return list of column descriptions: [{ "name": str, "data_type": str, "is_nullable': bool, "is_pk": bool }]

Source code in dataedit/helper.py

def get_column_description(schema, table):
    """Return list of column descriptions:
    [{
       "name": str,
       "data_type": str,
       "is_nullable': bool,
       "is_pk": bool
    }]

    """

    def get_datatype_str(column_def):
        """get single string sql type definition.

        We want the data type definition to be a simple string, e.g. decimal(10, 6)
        or varchar(128), so we need to combine the various fields
        (type, numeric_precision, numeric_scale, ...)
        """
        # for reverse validation, see also api.parser.parse_type(dt_string)
        dt = column_def["data_type"].lower()
        precisions = None
        if dt.startswith("character"):
            if dt == "character varying":
                dt = "varchar"
            else:
                dt = "char"
            precisions = [column_def["character_maximum_length"]]
        elif dt.endswith(" without time zone"):  # this is the default
            dt = dt.replace(" without time zone", "")
        elif re.match("(numeric|decimal)", dt):
            precisions = [column_def["numeric_precision"], column_def["numeric_scale"]]
        elif dt == "interval":
            precisions = [column_def["interval_precision"]]
        elif re.match(".*int", dt) and re.match(
            "nextval", column_def.get("column_default") or ""
        ):
            # dt = dt.replace('int', 'serial')
            pass
        elif dt.startswith("double"):
            dt = "float"
        if precisions:  # remove None
            precisions = [x for x in precisions if x is not None]
        if precisions:
            dt += "(%s)" % ", ".join(str(x) for x in precisions)
        return dt

    def get_pk_fields(constraints):
        """Get the column names that make up the primary key
        from the constraints definitions.

        NOTE: Currently, the wizard to create tables only supports
            single fields primary keys (which is advisable anyways)
        """
        pk_fields = []
        for _name, constraint in constraints.items():
            if constraint.get("constraint_type") == "PRIMARY KEY":
                m = re.match(
                    r"PRIMARY KEY[ ]*\(([^)]+)", constraint.get("definition") or ""
                )
                if m:
                    # "f1, f2" -> ["f1", "f2"]
                    pk_fields = [x.strip() for x in m.groups()[0].split(",")]
        return pk_fields

    _columns = actions.describe_columns(schema, table)
    _constraints = actions.describe_constraints(schema, table)
    pk_fields = get_pk_fields(_constraints)
    # order by ordinal_position
    columns = []
    for name, col in sorted(
        _columns.items(), key=lambda kv: int(kv[1]["ordinal_position"])
    ):
        columns.append(
            {
                "name": name,
                "data_type": get_datatype_str(col),
                "is_nullable": col["is_nullable"],
                "is_pk": name in pk_fields,
                "unit": None,
                "description": None,
            }
        )
    return columns

`get_review_for_key(key, review_data)` #

Retrieve the review for a specific key from the review data.

Parameters:

Name	Type	Description	Default
`key`	`str`	The key for which to retrieve the review. review_data (dict): The review data containing reviews for various keys.	required

Returns:

Name	Type	Description
`Any`		The new value associated with the specified key in the review data, or None if the key is not found.

Source code in dataedit/helper.py

def get_review_for_key(key, review_data):
    """
    Retrieve the review for a specific key from the review data.

    Args:
        key (str): The key for which to retrieve the review.
            review_data (dict): The review data containing
            reviews for various keys.

    Returns:
        Any: The new value associated with the specified key
            in the review data, or None if the key is not found.
    """

    for review in review_data["reviewData"]["reviews"]:
        if review["key"] == key:
            return review["fieldReview"].get("newValue", None)
    return None

`merge_field_reviews(current_json, new_json)` #

Merge reviews from contributors and reviewers into a single JSON object.

Parameters:

Name	Type	Description	Default
`current_json`	`dict`	The current JSON object containing reviewer's reviews.	required
`new_json`	`dict`	The new JSON object containing contributor's reviews.	required

Returns:

Name	Type	Description
`dict`		The merged JSON object containing both contributor's and reviewer's reviews.

Note

If the same key is present in both the contributor's and reviewer's reviews, the function will merge the field evaluations. Otherwise, it will create a new entry in the Review-Dict.

Source code in dataedit/helper.py

def merge_field_reviews(current_json, new_json):
    """
    Merge reviews from contributors and reviewers into a single JSON object.

    Args:
        current_json (dict): The current JSON object containing
            reviewer's reviews.
        new_json (dict): The new JSON object containing contributor's reviews.

    Returns:
        dict: The merged JSON object containing both contributor's and
            reviewer's reviews.

    Note:
        If the same key is present in both the contributor's and
            reviewer's reviews, the function will merge the field
            evaluations. Otherwise, it will create a new entry in
            the Review-Dict.
    """
    merged_json = new_json.copy()
    review_dict = {}

    for contrib_review in merged_json["reviews"]:
        category = contrib_review["category"]
        key = contrib_review["key"]
        review_dict[(category, key)] = contrib_review["fieldReview"]

    for reviewer_review in current_json["reviews"]:
        category = reviewer_review["category"]
        key = reviewer_review["key"]

        if (category, key) in review_dict:
            # Add field evaluations to the existing entry
            existing_field_review = review_dict[(category, key)]
            if isinstance(existing_field_review, dict):
                existing_field_review = [existing_field_review]
            if isinstance(reviewer_review["fieldReview"], dict):
                reviewer_review["fieldReview"] = [reviewer_review["fieldReview"]]
            merged_field_review = existing_field_review + reviewer_review["fieldReview"]
            review_dict[(category, key)] = merged_field_review
        else:
            # Create new entry in Review-Dict
            review_dict[(category, key)] = reviewer_review["fieldReview"]

    # Insert updated field scores back into the JSON
    merged_json["reviews"] = [
        {"category": category, "key": key, "fieldReview": review_dict[(category, key)]}
        for category, key in review_dict
    ]

    return merged_json

`recursive_update(metadata, review_data)` #

Recursively updates metadata with new values from review_data, skipping or removing fields with status 'rejected'.

Args: metadata (dict): The original metadata dictionary to update. review_data (dict): The review data containing the new values for various keys.

Note: The function iterates through the review data and for each key updates the corresponding value in metadata if the new value is present and is not an empty string, and if the field status is not 'rejected'.

Source code in dataedit/helper.py

def recursive_update(metadata, review_data):
    """
    Recursively updates metadata with new values from review_data,
    skipping or removing fields with status 'rejected'.

    Args:
    metadata (dict): The original metadata dictionary to update.
    review_data (dict): The review data containing the new values
    for various keys.

    Note:
    The function iterates through the review data and for each key
    updates the corresponding value in metadata if the new value is
    present and is not an empty string, and if the field status is
    not 'rejected'.
    """

    def delete_nested_field(data, keys):
        """
        Removes a nested field from a dictionary based on a list of keys.

        Args:
            data (dict or list): The dictionary or list from which
            to remove the field.
            keys (list): A list of keys pointing to the field to remove.
        """
        for key in keys[:-1]:
            if isinstance(data, list):
                key = int(key)
            data = data.get(key) if isinstance(data, dict) else data[key]

        last_key = keys[-1]
        if isinstance(data, list) and last_key.isdigit():
            index = int(last_key)
            if 0 <= index < len(data):
                data.pop(index)
        elif isinstance(data, dict):
            data.pop(last_key, None)

    for review_key in review_data["reviewData"]["reviews"]:
        keys = review_key["key"].split(".")

        field_review = review_key.get("fieldReview")
        if isinstance(field_review, list):
            field_rejected = False
            for fr in field_review:
                state = fr.get("state")
                if state == "rejected":
                    # If a field is rejected, delete it and move on to the next one.
                    delete_nested_field(metadata, keys)
                    field_rejected = True
                    break
            if field_rejected:
                continue

            # If the field is not rejected, apply the new value
            for fr in field_review:
                new_value = fr.get("newValue", None)
                if new_value is not None and new_value != "":
                    set_nested_value(metadata, keys, new_value)

        elif isinstance(field_review, dict):
            state = field_review.get("state")
            if state == "rejected":
                # If a field is rejected, delete it and move on to the next one.
                delete_nested_field(metadata, keys)
                continue

            # If the field is not rejected, apply the new value
            new_value = field_review.get("newValue", None)
            if new_value is not None and new_value != "":
                set_nested_value(metadata, keys, new_value)

    return metadata

`set_nested_value(metadata, keys, value)` #

Set a nested value in a dictionary given a sequence of keys.

Parameters:

Name	Type	Description	Default
`metadata`	`dict`	The dictionary in which to set the value.	required
`keys`	`list`	A list of keys representing the path to the nested value.	required
`value`	`Any`	The value to set.	required

Note

The function navigates through the dictionary using the keys and sets the value at the position indicated by the last key in the list.

Source code in dataedit/helper.py

def set_nested_value(metadata, keys, value):
    """
    Set a nested value in a dictionary given a sequence of keys.

    Args:
        metadata (dict): The dictionary in which to set the value.
        keys (list): A list of keys representing the path to the nested value.
        value (Any): The value to set.

    Note:
        The function navigates through the dictionary using the keys
        and sets the value at the position indicated by the last key in the list.
    """

    for key in keys[:-1]:
        if key.isdigit():
            key = int(key)
        metadata = metadata[key]
    last_key = keys[-1]
    if last_key.isdigit():
        last_key = int(last_key)
    metadata[last_key] = value

`update_keywords_from_tags(table, schema)` #

synchronize keywords in metadata with tags

Source code in dataedit/helper.py

def update_keywords_from_tags(table: Table, schema: str) -> None:
    """synchronize keywords in metadata with tags"""

    metadata = table.oemetadata or {"resources": [{}]}
    keywords = [tag.name_normalized for tag in table.tags.all()]
    metadata["resources"][0]["keywords"] = keywords

    actions.set_table_metadata(table=table.name, metadata=metadata)

Metadata Functions#

Provide functionality that is related to retrieving and updating the oemetadata resource from the database. The oemetadata is the object of a review.

Save Metadata to Database#

Save updated metadata for a specific table in the OEP database.

Parameters:

Name	Type	Description	Default
`schema`	`str`	The name of the OEP schema.	required
`table`	`str`	The name of the table in the OEP schema.	required
`updated_metadata`	`dict`	The updated metadata dictionary.	required

Note

This function loads the table object from the database, updates its metadata field, and then saves the updated table object back to the database.

Source code in dataedit/metadata/__init__.py

def save_metadata_to_db(schema: str, table: str, updated_metadata):
    """
    Save updated metadata for a specific table in the OEP database.

    Args:
        schema (str): The name of the OEP schema.
        table (str): The name of the table in the OEP schema.
        updated_metadata (dict): The updated metadata dictionary.

    Note:
        This function loads the table object from the database,
        updates its metadata field, and then saves the updated
        table object back to the database.
    """

    # Load the table object
    table_obj = Table.load(name=table)

    # Update the oemetadata field
    table_obj.oemetadata = updated_metadata

    # Save the updated table object
    table_obj.save()

Load Metadata from Database#

Load metadata for a specific table from the OEP database.

Parameters:

Name	Type	Description	Default
`table`	`str`	The name of the table in the OEP schema.	required

Returns:

Name	Type	Description
`dict`	`dict`	The loaded metadata dictionary.

Note

The function currently loads metadata from the Table.oemetadata field. There is a consideration to change this function to use a different approach or keep the old functionality (TODO).

Source code in dataedit/metadata/__init__.py

def load_metadata_from_db(table: str) -> dict:
    """
    Load metadata for a specific table from the OEP database.

    Args:
        table (str): The name of the table in the OEP schema.

    Returns:
        dict: The loaded metadata dictionary.

    Note:
        The function currently loads metadata from the Table.oemetadata field.
        There is a consideration to change this function to use a different approach
        or keep the old functionality (TODO).
    """

    metadata = Table.load(name=table).oemetadata
    if not metadata:
        metadata = OEMETADATA_V20_TEMPLATE
    return metadata

Models#

Django models.

PeerReview#

The model of the Open Peer Review defines what data is stored in the django database about each existing review. Next to the review itself it stores additional data about the the reviewer and contributor user and more. It is used in the PeerReviewManger.

Note

This model also provides functionality that is directly related to the model. It is up to discussion if we want to keep the functionality inside the model.

Bases: Model

Represents a peer review in the database.

Attributes:

Name	Type	Description
`table`	`CharField`	Name of the table being reviewed.
`reviewer`	`ForeignKey`	The user who reviews.
`contributor`	`ForeignKey`	The user who contributes.
`is_finished`	`BooleanField`	Whether the review is finished.
`date_started`	`DateTimeField`	When the review started.
`date_submitted`	`DateTimeField`	When the review was submitted.
`date_finished`	`DateTimeField`	When the review finished.
`review`	`JSONField`	The review data in JSON format.

Source code in dataedit/models.py

class PeerReview(models.Model):
    """
    Represents a peer review in the database.

    Attributes:
        table (CharField): Name of the table being reviewed.
        reviewer (ForeignKey): The user who reviews.
        contributor (ForeignKey): The user who contributes.
        is_finished (BooleanField): Whether the review is finished.
        date_started (DateTimeField): When the review started.
        date_submitted (DateTimeField): When the review was submitted.
        date_finished (DateTimeField): When the review finished.
        review (JSONField): The review data in JSON format.
    """

    table = CharField(max_length=1000, null=False)
    reviewer = ForeignKey(
        "login.myuser", on_delete=models.CASCADE, related_name="reviewed_by", null=True
    )
    contributor = ForeignKey(
        "login.myuser",
        on_delete=models.CASCADE,
        related_name="review_received",
        null=True,
    )
    is_finished = BooleanField(null=False, default=False)
    date_started = DateTimeField(max_length=1000, null=False, default=timezone.now)
    date_submitted = DateTimeField(max_length=1000, null=True, default=None)
    date_finished = DateTimeField(max_length=1000, null=True, default=None)
    review = JSONField(null=True)
    # TODO: Maybe oemetadata should be stored in a separate table and imported
    # via FK here / change also for Tables model
    oemetadata = JSONField(null=False, default=dict)

    # laden
    @classmethod
    def load(cls, table: str) -> Union["PeerReview", None]:
        """
        Load the current reviewer user.
        The current review is review is determened by the latest date started.

        Args:
            table (string): Table name

        Returns:
            opr (PeerReview): PeerReview object related to the latest
            date started.
        """
        opr = PeerReview.objects.filter(table=table).order_by("-date_started").first()
        return opr

    # TODO: CAUTION unfinished work ... fix: includes all id´s and not just the
    # related ones (reviews on same table) .. procedures false results
    def get_prev_and_next_reviews(self, table: str):
        """
        Sets the prev_review and next_review fields based on the date_started
        field of the PeerReview objects associated with the same table.
        """
        # Get all the PeerReview objects associated with the same table name
        peer_reviews = PeerReview.objects.filter(table=table).order_by("date_started")

        current_index = None
        for index, review in enumerate(peer_reviews):
            if review.pk == self.pk:
                current_index = index
                break

        prev_review = None
        next_review = None

        if current_index is not None:
            if current_index > 0:
                prev_review = peer_reviews[current_index - 1]

            if current_index < len(peer_reviews) - 1:
                next_review = peer_reviews[current_index + 1]

        return prev_review, next_review

    def save(self, *args, **kwargs):
        review_type = kwargs.pop("review_type", None)
        pm_new = None

        if not self.contributor == self.reviewer:
            super().save(*args, **kwargs)
            # TODO: This causes errors if review list ist empty

            if review_type == "save":
                pm_new = PeerReviewManager(
                    opr=self, status=ReviewDataStatus.SAVED.value
                )

            elif review_type == "submit":
                result = self.set_version_of_metadata_for_review(table=self.table)
                if result[0]:
                    logging.info(result[1])
                elif result[0] is False:
                    logging.info(result[1])

                pm_new = PeerReviewManager(
                    opr=self, status=ReviewDataStatus.SUBMITTED.value
                )
                pm_new.set_next_reviewer()

            elif review_type == "finished":
                result = self.set_version_of_metadata_for_review(table=self.table)
                if result[0]:
                    logging.info(result[1])
                elif result[0] is False:
                    logging.info(result[1])

                pm_new = PeerReviewManager(
                    opr=self, status=ReviewDataStatus.FINISHED.value
                )
                self.is_finished = True
                self.date_finished = timezone.now()
                super().save(*args, **kwargs)

            if pm_new:
                pm_new.save()

        else:
            raise ValidationError("Contributor and reviewer cannot be the same.")

    def delete(self, *args, **kwargs):
        """
        Custom delete method to remove related PeerReviewManager entries.
        """
        # Remove related records in PeerReviewManager
        PeerReviewManager.objects.filter(opr=self).delete()

        super().delete(*args, **kwargs)

    def update(self, *args, **kwargs):
        """
        Update the peer review if the latest peer review is not finished yet
        but either saved or submitted.

        """

        review_type = kwargs.pop("review_type", None)
        if not self.contributor == self.reviewer:
            current_pm = PeerReviewManager.load(opr=self)
            if review_type == "save":
                current_pm.status = ReviewDataStatus.SAVED.value
            elif review_type == "submit":
                current_pm.status = ReviewDataStatus.SUBMITTED.value
                current_pm.set_next_reviewer()
            elif review_type == "finished":
                self.is_finished = True
                self.date_finished = timezone.now()
                current_pm.status = ReviewDataStatus.FINISHED.value

            # update peere review manager related to this peer review entry
            current_pm.save()
            super().save(*args, **kwargs)
        else:
            raise ValidationError("Contributor and reviewer cannot be the same.")

    def set_version_of_metadata_for_review(self, table: str, *args, **kwargs):
        """
        Once the peer review is started, we save the current version of the
        oemetadata that is present on the table to the peer review instance
        to be able to do the review to a fixed state of the metadata.

        A started review means a reviewer saves / submits or finishes (in case
        the review is completed in one go) a review.

        Args:
            table (str): Table name

        Returns:
            State (tuple): Bool value that indicates weather there is already
            a version of oemetadata available for this review & readable
            status message.
        """
        table_oemetdata = Table.load(name=table).oemetadata

        if self.oemetadata is None:
            self.oemetadata = table_oemetdata
            super().save(*args, **kwargs)

            return (
                True,
                f"Set current version of table's: '{table}' " "oemetadata for review.",
            )

        return (
            False,
            f"This tables (name: {table}) review "
            "already got a version of oemetadata.",
        )

    def update_all_table_peer_reviews_after_table_moved(self, *args, topic, **kwargs):
        if isinstance(self.review, str):
            review_data = json.loads(self.review)
        else:
            review_data = self.review

        review_data["topic"] = topic

        self.review = review_data

        super().save(*args, **kwargs)

    @property
    def days_open(self):
        if self.date_started is None:
            return None  # Review has not started yet
        elif self.is_finished:
            return (self.date_finished - self.date_started).days  # Review has finished
        else:
            return (timezone.now() - self.date_started).days  # Review is still open

`delete(*args, **kwargs)` #

Custom delete method to remove related PeerReviewManager entries.

Source code in dataedit/models.py

def delete(self, *args, **kwargs):
    """
    Custom delete method to remove related PeerReviewManager entries.
    """
    # Remove related records in PeerReviewManager
    PeerReviewManager.objects.filter(opr=self).delete()

    super().delete(*args, **kwargs)

`get_prev_and_next_reviews(table)` #

Sets the prev_review and next_review fields based on the date_started field of the PeerReview objects associated with the same table.

Source code in dataedit/models.py

def get_prev_and_next_reviews(self, table: str):
    """
    Sets the prev_review and next_review fields based on the date_started
    field of the PeerReview objects associated with the same table.
    """
    # Get all the PeerReview objects associated with the same table name
    peer_reviews = PeerReview.objects.filter(table=table).order_by("date_started")

    current_index = None
    for index, review in enumerate(peer_reviews):
        if review.pk == self.pk:
            current_index = index
            break

    prev_review = None
    next_review = None

    if current_index is not None:
        if current_index > 0:
            prev_review = peer_reviews[current_index - 1]

        if current_index < len(peer_reviews) - 1:
            next_review = peer_reviews[current_index + 1]

    return prev_review, next_review

`load(table)` `classmethod` #

Load the current reviewer user. The current review is review is determened by the latest date started.

Parameters:

Name	Type	Description	Default
`table`	`string`	Table name	required

Returns:

Name	Type	Description
`opr`	`PeerReview`	PeerReview object related to the latest
	`Union[PeerReview, None]`	date started.

Source code in dataedit/models.py

@classmethod
def load(cls, table: str) -> Union["PeerReview", None]:
    """
    Load the current reviewer user.
    The current review is review is determened by the latest date started.

    Args:
        table (string): Table name

    Returns:
        opr (PeerReview): PeerReview object related to the latest
        date started.
    """
    opr = PeerReview.objects.filter(table=table).order_by("-date_started").first()
    return opr

`set_version_of_metadata_for_review(table, *args, **kwargs)` #

Once the peer review is started, we save the current version of the oemetadata that is present on the table to the peer review instance to be able to do the review to a fixed state of the metadata.

A started review means a reviewer saves / submits or finishes (in case the review is completed in one go) a review.

Parameters:

Name	Type	Description	Default
`table`	`str`	Table name	required

Returns:

Name	Type	Description
`State`	`tuple`	Bool value that indicates weather there is already
		a version of oemetadata available for this review & readable
		status message.

Source code in dataedit/models.py

def set_version_of_metadata_for_review(self, table: str, *args, **kwargs):
    """
    Once the peer review is started, we save the current version of the
    oemetadata that is present on the table to the peer review instance
    to be able to do the review to a fixed state of the metadata.

    A started review means a reviewer saves / submits or finishes (in case
    the review is completed in one go) a review.

    Args:
        table (str): Table name

    Returns:
        State (tuple): Bool value that indicates weather there is already
        a version of oemetadata available for this review & readable
        status message.
    """
    table_oemetdata = Table.load(name=table).oemetadata

    if self.oemetadata is None:
        self.oemetadata = table_oemetdata
        super().save(*args, **kwargs)

        return (
            True,
            f"Set current version of table's: '{table}' " "oemetadata for review.",
        )

    return (
        False,
        f"This tables (name: {table}) review "
        "already got a version of oemetadata.",
    )

`update(*args, **kwargs)` #

Update the peer review if the latest peer review is not finished yet but either saved or submitted.

Source code in dataedit/models.py

def update(self, *args, **kwargs):
    """
    Update the peer review if the latest peer review is not finished yet
    but either saved or submitted.

    """

    review_type = kwargs.pop("review_type", None)
    if not self.contributor == self.reviewer:
        current_pm = PeerReviewManager.load(opr=self)
        if review_type == "save":
            current_pm.status = ReviewDataStatus.SAVED.value
        elif review_type == "submit":
            current_pm.status = ReviewDataStatus.SUBMITTED.value
            current_pm.set_next_reviewer()
        elif review_type == "finished":
            self.is_finished = True
            self.date_finished = timezone.now()
            current_pm.status = ReviewDataStatus.FINISHED.value

        # update peere review manager related to this peer review entry
        current_pm.save()
        super().save(*args, **kwargs)
    else:
        raise ValidationError("Contributor and reviewer cannot be the same.")

PeerReviewManager#

The Manager is introduced to be able to store additional information about the peer review process and separate it from the PeerReview model. The process is started by submitting a review and the manager maintains the order of which user has to take the next action to be able to hold and activate the process.

Note

This model also provides functionality that is directly related to the model. It is up to discussion if we want to keep the functionality inside the model.

Bases: Model

Manages peer review processes.

This model handles the 1:n relation between table and open peer reviews. It tracks the days open for the peer review and its state. It determines who is next in the process between reviewer and contributor. It provides information about the previous and next review. It offers several methods that provide generic filters for the peer reviews.

Attributes:

Name	Type	Description
`opr`	`ForeignKey`	The associated peer review.
`current_reviewer`	`CharField`	The current reviewer.
`status`	`CharField`	The current status of the review.
`is_open_since`	`CharField`	How long the review has been open.
`prev_review`	`ForeignKey`	The previous review in the process.
`next_review`	`ForeignKey`	The next review in the process.

Source code in dataedit/models.py

class PeerReviewManager(models.Model):
    """
    Manages peer review processes.

    This model handles the 1:n relation between table and open peer reviews.
    It tracks the days open for the peer review and its state.
    It determines who is next in the process between reviewer and contributor.
    It provides information about the previous and next review.
    It offers several methods that provide generic filters for the peer reviews.

    Attributes:
        opr (ForeignKey): The associated peer review.
        current_reviewer (CharField): The current reviewer.
        status (CharField): The current status of the review.
        is_open_since (CharField): How long the review has been open.
        prev_review (ForeignKey): The previous review in the process.
        next_review (ForeignKey): The next review in the process.
    """

    REVIEW_STATUS = [(status.value, status.name) for status in ReviewDataStatus]
    REVIEWER_CHOICES = [(choice.value, choice.name) for choice in Reviewer]

    opr = ForeignKey(
        PeerReview, on_delete=models.CASCADE, related_name="review_id", null=False
    )
    current_reviewer = models.CharField(
        choices=REVIEWER_CHOICES, max_length=20, default=Reviewer.REVIEWER.value
    )
    status = models.CharField(
        choices=REVIEW_STATUS, max_length=10, default=ReviewDataStatus.SAVED.value
    )
    is_open_since = models.CharField(null=True, max_length=10)
    prev_review = ForeignKey(
        PeerReview,
        on_delete=models.CASCADE,
        related_name="prev_review",
        null=True,
        default=None,
    )  # TODO: add logic
    next_review = ForeignKey(
        PeerReview,
        on_delete=models.CASCADE,
        related_name="next_review",
        null=True,
        default=None,
    )  # TODO: add logic

    @classmethod
    def load(cls, opr):
        """
        Load the peer review manager associated with the given peer review.

        Args:
            opr (PeerReview): The peer review.

        Returns:
            PeerReviewManager: The peer review manager.
        """
        peer_review_manager = PeerReviewManager.objects.get(opr=opr)
        return peer_review_manager

    def save(self, *args, **kwargs):
        """
        Override the save method to perform additional logic
        before saving the peer review manager.
        """
        # Set is_open_since field if it is None
        if self.is_open_since is None:
            # Get the associated PeerReview instance
            peer_review = self.opr

            # Set is_open_since based on the days_open property of the
            # PeerReview instance
            days_open = peer_review.days_open
            if days_open is not None:
                self.is_open_since = str(days_open)
        # Call the parent class's save method to save the PeerReviewManager instance
        super().save(*args, **kwargs)

    @classmethod
    def update_open_since(cls, opr=None, *args, **kwargs):
        """
        Update the "is_open_since" field of the peer review manager.

        Args:
            opr (PeerReview): The peer review.
            If None, use the peer review associated with the manager.

        """
        if opr is not None:
            peer_review = PeerReviewManager.objects.get(opr=opr)
        else:
            peer_review = cls.opr

        days_open = peer_review.opr.days_open
        peer_review.is_open_since = str(days_open)

        # Call the parent class's save method to save the PeerReviewManager instance
        peer_review.save(*args, **kwargs)

    def set_next_reviewer(self):
        """
        Set the order on which peer will be required to perform a action to
        continue with the process.
        """
        # TODO:check for user identifies as ...
        if self.current_reviewer == Reviewer.REVIEWER.value:
            self.current_reviewer = Reviewer.CONTRIBUTOR.value
        else:
            self.current_reviewer = Reviewer.REVIEWER.value
        self.save()

    def whos_turn(self):
        """
        Get the user and role (contributor or reviewer) whose turn it is.

        Returns:
            Tuple[str, User]: The role and user.
        """
        role, result = None, None
        peer_review = self.opr
        if self.current_reviewer == Reviewer.REVIEWER.value:
            role = Reviewer.REVIEWER.value
            result = peer_review.reviewer
        else:
            role = Reviewer.CONTRIBUTOR.value
            result = peer_review.contributor

        return role, result

    @staticmethod
    def load_contributor(table: str):
        """
        Get the contributor for the table a review is started.

        Args:
            table (str): Table name.

        Returns:
            User: The contributor user.
        """
        current_table = Table.load(name=table)
        try:
            table_holder = (
                current_table.userpermission_set.filter(table=current_table.pk)
                .first()
                .holder
            )
        except AttributeError:
            table_holder = None
        return table_holder

    @staticmethod
    def load_reviewer(table: str):
        """
        Get the reviewer for the table a review is started.

        Args:
            table (str): Table name.

        Returns:
            User: The reviewer user.
        """
        current_review = PeerReview.load(table=table)
        if current_review and hasattr(current_review, "reviewer"):
            return current_review.reviewer
        else:
            return None

    @staticmethod
    def filter_opr_by_reviewer(reviewer_user):
        """
        Filter peer reviews by reviewer, excluding those with current peer
        is contributor and the data status "SAVED" in the peer review manager.

        Args:
            reviewer_user (User): The reviewer user.

        Returns:
            QuerySet: Filtered peer reviews.
        """
        return PeerReview.objects.filter(reviewer=reviewer_user).exclude(
            review_id__current_reviewer=Reviewer.CONTRIBUTOR.value,
            review_id__status=ReviewDataStatus.SAVED.value,
        )

    @staticmethod
    def filter_latest_open_opr_by_reviewer(reviewer_user):
        """
        Get the last open peer review for the given contributor.

        Args:
            contributor_user (User): The contributor user.

        Returns:
            PeerReview: Last open peer review or None if not found.
        """
        try:
            return (
                PeerReview.objects.filter(reviewer=reviewer_user, is_finished=False)
                .exclude(
                    review_id__current_reviewer=Reviewer.CONTRIBUTOR.value,
                    review_id__status=ReviewDataStatus.SAVED.value,
                )
                .latest("date_started")
            )
        except PeerReview.DoesNotExist:
            return None

    @staticmethod
    def filter_opr_by_contributor(contributor_user):
        """
        Filter peer reviews by contributor, excluding those with current peer
        is reviewer and the data status "SAVED" in the peer review manager.

        Args:
            contributor_user (User): The contributor user.

        Returns:
            QuerySet: Filtered peer reviews.
        """

        return PeerReview.objects.filter(contributor=contributor_user).exclude(
            review_id__current_reviewer=Reviewer.REVIEWER.value,
            review_id__status=ReviewDataStatus.SAVED.value,
        )

    @staticmethod
    def filter_latest_open_opr_by_contributor(contributor_user):
        """
        Get the last open peer review for the given contributor.

        Args:
            contributor_user (User): The contributor user.

        Returns:
            PeerReview: Last open peer review or None if not found.
        """
        try:
            return (
                PeerReview.objects.filter(
                    contributor=contributor_user, is_finished=False
                )
                .exclude(
                    review_id__current_reviewer=Reviewer.REVIEWER.value,
                    review_id__status=ReviewDataStatus.SAVED.value,
                )
                .latest("date_started")
            )
        except PeerReview.DoesNotExist:
            return None

    @staticmethod
    def filter_opr_by_table(table: str) -> QuerySet[PeerReview]:
        """
        Filter peer reviews by table.

        Args:
            table (str): Table name.

        Returns:
            QuerySet: Filtered peer reviews.
        """
        return PeerReview.objects.filter(table=table)

    @staticmethod
    def get_opr_by_id(opr_id) -> PeerReview:
        return PeerReview.objects.get(id=opr_id)

`filter_latest_open_opr_by_contributor(contributor_user)` `staticmethod` #

Get the last open peer review for the given contributor.

Parameters:

Name	Type	Description	Default
`contributor_user`	`User`	The contributor user.	required

Returns:

Name	Type	Description
`PeerReview`		Last open peer review or None if not found.

Source code in dataedit/models.py

@staticmethod
def filter_latest_open_opr_by_contributor(contributor_user):
    """
    Get the last open peer review for the given contributor.

    Args:
        contributor_user (User): The contributor user.

    Returns:
        PeerReview: Last open peer review or None if not found.
    """
    try:
        return (
            PeerReview.objects.filter(
                contributor=contributor_user, is_finished=False
            )
            .exclude(
                review_id__current_reviewer=Reviewer.REVIEWER.value,
                review_id__status=ReviewDataStatus.SAVED.value,
            )
            .latest("date_started")
        )
    except PeerReview.DoesNotExist:
        return None

`filter_latest_open_opr_by_reviewer(reviewer_user)` `staticmethod` #

Get the last open peer review for the given contributor.

Parameters:

Name	Type	Description	Default
`contributor_user`	`User`	The contributor user.	required

Returns:

Name	Type	Description
`PeerReview`		Last open peer review or None if not found.

Source code in dataedit/models.py

@staticmethod
def filter_latest_open_opr_by_reviewer(reviewer_user):
    """
    Get the last open peer review for the given contributor.

    Args:
        contributor_user (User): The contributor user.

    Returns:
        PeerReview: Last open peer review or None if not found.
    """
    try:
        return (
            PeerReview.objects.filter(reviewer=reviewer_user, is_finished=False)
            .exclude(
                review_id__current_reviewer=Reviewer.CONTRIBUTOR.value,
                review_id__status=ReviewDataStatus.SAVED.value,
            )
            .latest("date_started")
        )
    except PeerReview.DoesNotExist:
        return None

`filter_opr_by_contributor(contributor_user)` `staticmethod` #

Filter peer reviews by contributor, excluding those with current peer is reviewer and the data status "SAVED" in the peer review manager.

Parameters:

Name	Type	Description	Default
`contributor_user`	`User`	The contributor user.	required

Returns:

Name	Type	Description
`QuerySet`		Filtered peer reviews.

Source code in dataedit/models.py

@staticmethod
def filter_opr_by_contributor(contributor_user):
    """
    Filter peer reviews by contributor, excluding those with current peer
    is reviewer and the data status "SAVED" in the peer review manager.

    Args:
        contributor_user (User): The contributor user.

    Returns:
        QuerySet: Filtered peer reviews.
    """

    return PeerReview.objects.filter(contributor=contributor_user).exclude(
        review_id__current_reviewer=Reviewer.REVIEWER.value,
        review_id__status=ReviewDataStatus.SAVED.value,
    )

`filter_opr_by_reviewer(reviewer_user)` `staticmethod` #

Filter peer reviews by reviewer, excluding those with current peer is contributor and the data status "SAVED" in the peer review manager.

Parameters:

Name	Type	Description	Default
`reviewer_user`	`User`	The reviewer user.	required

Returns:

Name	Type	Description
`QuerySet`		Filtered peer reviews.

Source code in dataedit/models.py

@staticmethod
def filter_opr_by_reviewer(reviewer_user):
    """
    Filter peer reviews by reviewer, excluding those with current peer
    is contributor and the data status "SAVED" in the peer review manager.

    Args:
        reviewer_user (User): The reviewer user.

    Returns:
        QuerySet: Filtered peer reviews.
    """
    return PeerReview.objects.filter(reviewer=reviewer_user).exclude(
        review_id__current_reviewer=Reviewer.CONTRIBUTOR.value,
        review_id__status=ReviewDataStatus.SAVED.value,
    )

`filter_opr_by_table(table)` `staticmethod` #

Filter peer reviews by table.

Parameters:

Name	Type	Description	Default
`table`	`str`	Table name.	required

Returns:

Name	Type	Description
`QuerySet`	`QuerySet[PeerReview]`	Filtered peer reviews.

Source code in dataedit/models.py

@staticmethod
def filter_opr_by_table(table: str) -> QuerySet[PeerReview]:
    """
    Filter peer reviews by table.

    Args:
        table (str): Table name.

    Returns:
        QuerySet: Filtered peer reviews.
    """
    return PeerReview.objects.filter(table=table)

`load(opr)` `classmethod` #

Load the peer review manager associated with the given peer review.

Parameters:

Name	Type	Description	Default
`opr`	`PeerReview`	The peer review.	required

Returns:

Name	Type	Description
`PeerReviewManager`		The peer review manager.

Source code in dataedit/models.py

@classmethod
def load(cls, opr):
    """
    Load the peer review manager associated with the given peer review.

    Args:
        opr (PeerReview): The peer review.

    Returns:
        PeerReviewManager: The peer review manager.
    """
    peer_review_manager = PeerReviewManager.objects.get(opr=opr)
    return peer_review_manager

`load_contributor(table)` `staticmethod` #

Get the contributor for the table a review is started.

Parameters:

Name	Type	Description	Default
`table`	`str`	Table name.	required

Returns:

Name	Type	Description
`User`		The contributor user.

Source code in dataedit/models.py

@staticmethod
def load_contributor(table: str):
    """
    Get the contributor for the table a review is started.

    Args:
        table (str): Table name.

    Returns:
        User: The contributor user.
    """
    current_table = Table.load(name=table)
    try:
        table_holder = (
            current_table.userpermission_set.filter(table=current_table.pk)
            .first()
            .holder
        )
    except AttributeError:
        table_holder = None
    return table_holder

`load_reviewer(table)` `staticmethod` #

Get the reviewer for the table a review is started.

Parameters:

Name	Type	Description	Default
`table`	`str`	Table name.	required

Returns:

Name	Type	Description
`User`		The reviewer user.

Source code in dataedit/models.py

@staticmethod
def load_reviewer(table: str):
    """
    Get the reviewer for the table a review is started.

    Args:
        table (str): Table name.

    Returns:
        User: The reviewer user.
    """
    current_review = PeerReview.load(table=table)
    if current_review and hasattr(current_review, "reviewer"):
        return current_review.reviewer
    else:
        return None

`save(*args, **kwargs)` #

Override the save method to perform additional logic before saving the peer review manager.

Source code in dataedit/models.py

def save(self, *args, **kwargs):
    """
    Override the save method to perform additional logic
    before saving the peer review manager.
    """
    # Set is_open_since field if it is None
    if self.is_open_since is None:
        # Get the associated PeerReview instance
        peer_review = self.opr

        # Set is_open_since based on the days_open property of the
        # PeerReview instance
        days_open = peer_review.days_open
        if days_open is not None:
            self.is_open_since = str(days_open)
    # Call the parent class's save method to save the PeerReviewManager instance
    super().save(*args, **kwargs)

`set_next_reviewer()` #

Set the order on which peer will be required to perform a action to continue with the process.

Source code in dataedit/models.py

def set_next_reviewer(self):
    """
    Set the order on which peer will be required to perform a action to
    continue with the process.
    """
    # TODO:check for user identifies as ...
    if self.current_reviewer == Reviewer.REVIEWER.value:
        self.current_reviewer = Reviewer.CONTRIBUTOR.value
    else:
        self.current_reviewer = Reviewer.REVIEWER.value
    self.save()

`update_open_since(opr=None, *args, **kwargs)` `classmethod` #

Update the "is_open_since" field of the peer review manager.

Parameters:

Name	Type	Description	Default
`opr`	`PeerReview`	The peer review.	`None`

Source code in dataedit/models.py

@classmethod
def update_open_since(cls, opr=None, *args, **kwargs):
    """
    Update the "is_open_since" field of the peer review manager.

    Args:
        opr (PeerReview): The peer review.
        If None, use the peer review associated with the manager.

    """
    if opr is not None:
        peer_review = PeerReviewManager.objects.get(opr=opr)
    else:
        peer_review = cls.opr

    days_open = peer_review.opr.days_open
    peer_review.is_open_since = str(days_open)

    # Call the parent class's save method to save the PeerReviewManager instance
    peer_review.save(*args, **kwargs)

`whos_turn()` #

Get the user and role (contributor or reviewer) whose turn it is.

Returns:

Type	Description
	Tuple[str, User]: The role and user.

Source code in dataedit/models.py

def whos_turn(self):
    """
    Get the user and role (contributor or reviewer) whose turn it is.

    Returns:
        Tuple[str, User]: The role and user.
    """
    role, result = None, None
    peer_review = self.opr
    if self.current_reviewer == Reviewer.REVIEWER.value:
        role = Reviewer.REVIEWER.value
        result = peer_review.reviewer
    else:
        role = Reviewer.CONTRIBUTOR.value
        result = peer_review.contributor

    return role, result

Technical docs

Views#

PeerReviewView#

get(request, schema, table, review_id=None) #

get_all_field_descriptions(json_schema, prefix='') #

load_json(schema, table, review_id=None) #

load_json_schema() #

parse_keys(val, old='') #

post(request, schema, table, review_id=None) #

sort_in_category(schema, table, oemetadata) #

PeerRreviewContributorView#

get(request, schema, table, review_id) #

post(request, schema, table, review_id) #

Helper Functions#

add_tag(name, color) #

change_requests(schema, table) #

delete_peer_review(review_id) #

edit_tag(id, name, color) #

find_tables(topic_name=None, query_string=None, tag_ids=None) #

get_all_tags(schema_name=None, table_name=None) #

get_column_description(schema, table) #

get_review_for_key(key, review_data) #

merge_field_reviews(current_json, new_json) #

recursive_update(metadata, review_data) #

set_nested_value(metadata, keys, value) #

update_keywords_from_tags(table, schema) #

Metadata Functions#

Save Metadata to Database#

Load Metadata from Database#

Models#

PeerReview#

delete(*args, **kwargs) #

get_prev_and_next_reviews(table) #

load(table) classmethod #

set_version_of_metadata_for_review(table, *args, **kwargs) #

update(*args, **kwargs) #

PeerReviewManager#

filter_latest_open_opr_by_contributor(contributor_user) staticmethod #

filter_latest_open_opr_by_reviewer(reviewer_user) staticmethod #

filter_opr_by_contributor(contributor_user) staticmethod #

filter_opr_by_reviewer(reviewer_user) staticmethod #

filter_opr_by_table(table) staticmethod #

load(opr) classmethod #

load_contributor(table) staticmethod #

load_reviewer(table) staticmethod #

save(*args, **kwargs) #

set_next_reviewer() #

update_open_since(opr=None, *args, **kwargs) classmethod #

whos_turn() #

`get(request, schema, table, review_id=None)` #

`get_all_field_descriptions(json_schema, prefix='')` #

`load_json(schema, table, review_id=None)` #

`load_json_schema()` #

`parse_keys(val, old='')` #

`post(request, schema, table, review_id=None)` #

`sort_in_category(schema, table, oemetadata)` #

`get(request, schema, table, review_id)` #

`post(request, schema, table, review_id)` #

`add_tag(name, color)` #

`change_requests(schema, table)` #

`delete_peer_review(review_id)` #

`edit_tag(id, name, color)` #

`find_tables(topic_name=None, query_string=None, tag_ids=None)` #

`get_all_tags(schema_name=None, table_name=None)` #

`get_column_description(schema, table)` #

`get_review_for_key(key, review_data)` #

`merge_field_reviews(current_json, new_json)` #

`recursive_update(metadata, review_data)` #

`set_nested_value(metadata, keys, value)` #

`update_keywords_from_tags(table, schema)` #

`delete(*args, **kwargs)` #

`get_prev_and_next_reviews(table)` #

`load(table)` `classmethod` #

`set_version_of_metadata_for_review(table, *args, **kwargs)` #

`update(*args, **kwargs)` #

`filter_latest_open_opr_by_contributor(contributor_user)` `staticmethod` #

`filter_latest_open_opr_by_reviewer(reviewer_user)` `staticmethod` #

`filter_opr_by_contributor(contributor_user)` `staticmethod` #

`filter_opr_by_reviewer(reviewer_user)` `staticmethod` #

`filter_opr_by_table(table)` `staticmethod` #

`load(opr)` `classmethod` #

`load_contributor(table)` `staticmethod` #

`load_reviewer(table)` `staticmethod` #

`save(*args, **kwargs)` #

`set_next_reviewer()` #

`update_open_since(opr=None, *args, **kwargs)` `classmethod` #

`whos_turn()` #