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 PeerReviewView(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, 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.

        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(schema, table)
        elif review_id:
            opr = PeerReviewManager.filter_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_V160_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)}]
                # pass
            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, table, oemetadata):
        """
        Sorts the metadata of a table into categories and adds the value
        suggestion and comment that were added during the review, to facilitate
        Further processing easier.

        Note:
            The categories spatial & temporal are often combined during visualization.

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

        Returns:


        Examples:
            A return value can look like the below dictionary:

            >>>
            {
                "general": [
                    {
                    "field": "id",
                    "value": "http: //127.0.0.1:8000/dataedit/view/model_draft/test2",
                    "newValue": "",
                    "reviewer_suggestion": "",
                    "suggestion_comment": ""
                    }
                ],
                "spatial": [...],
                "temporal": [...],
                "source": [...],
                "license": [...],
            }

        """

        val = self.parse_keys(oemetadata)
        gen_key_list = []
        spatial_key_list = []
        temporal_key_list = []
        source_key_list = []
        license_key_list = []

        for i in val:
            fieldKey = list(i.values())[0]
            if fieldKey.split(".")[0] == "spatial":
                spatial_key_list.append(i)
            elif fieldKey.split(".")[0] == "temporal":
                temporal_key_list.append(i)
            elif fieldKey.split(".")[0] == "sources":
                source_key_list.append(i)
            elif fieldKey.split(".")[0] == "licenses":
                license_key_list.append(i)

            elif (
                fieldKey.split(".")[0] == "name"
                or fieldKey.split(".")[0] == "title"
                or fieldKey.split(".")[0] == "id"
                or fieldKey.split(".")[0] == "description"
                or fieldKey.split(".")[0] == "language"
                or fieldKey.split(".")[0] == "subject"
                or fieldKey.split(".")[0] == "keywords"
                or fieldKey.split(".")[0] == "publicationDate"
                or fieldKey.split(".")[0] == "context"
            ):
                gen_key_list.append(i)

        meta = {
            "general": gen_key_list,
            "spatial": spatial_key_list,
            "temporal": temporal_key_list,
            "source": source_key_list,
            "license": license_key_list,
        }

        return meta

    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", "example", "badge", "title"]
                ):
                    field_descriptions[key] = {}
                    if "description" in value:
                        field_descriptions[key]["description"] = value["description"]
                    if "example" in value:
                        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, schema, table, review_id=None):
        """
        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(schema, table)
        field_descriptions = self.get_all_field_descriptions(json_schema)

        # Check user permissions
        if not request.user.is_anonymous:
            level = request.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.filter_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,
        }
        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),
        }
        return render(request, "dataedit/opr_review.html", context=context_meta)

    def post(self, 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

        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 = {}
        if request.method == "POST":
            # 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.
            contributor = PeerReviewManager.load_contributor(schema, table)

            if contributor is not None:
                # Überprüfen, ob ein aktiver PeerReview existiert
                active_peer_review = PeerReview.load(schema=schema, 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(
                        schema=schema,
                        table=table,
                        is_finished=review_finished,
                        review=review_datamodel,
                        reviewer=request.user,
                        contributor=contributor,
                        oemetadata=load_metadata_from_db(schema=schema, 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 = request.user
                    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(schema=schema, table=table)
                review_table.set_is_reviewed()
                metadata = self.load_json(schema, table, review_id=review_id)

                recursive_update(metadata, review_data)

                save_metadata_to_db(schema, table, metadata)

                if active_peer_review:
                    # Update the oemetadata in the active PeerReview
                    active_peer_review.oemetadata = metadata
                    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`		Rendered HTML response.

Source code in dataedit/views.py

def get(self, request, schema, table, review_id=None):
    """
    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(schema, table)
    field_descriptions = self.get_all_field_descriptions(json_schema)

    # Check user permissions
    if not request.user.is_anonymous:
        level = request.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.filter_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,
    }
    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),
    }
    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", "example", "badge", "title"]
            ):
                field_descriptions[key] = {}
                if "description" in value:
                    field_descriptions[key]["description"] = value["description"]
                if "example" in value:
                    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, 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.

    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(schema, table)
    elif review_id:
        opr = PeerReviewManager.filter_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_V160_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)}]
            # pass
        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`		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, 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

    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 = {}
    if request.method == "POST":
        # 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.
        contributor = PeerReviewManager.load_contributor(schema, table)

        if contributor is not None:
            # Überprüfen, ob ein aktiver PeerReview existiert
            active_peer_review = PeerReview.load(schema=schema, 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(
                    schema=schema,
                    table=table,
                    is_finished=review_finished,
                    review=review_datamodel,
                    reviewer=request.user,
                    contributor=contributor,
                    oemetadata=load_metadata_from_db(schema=schema, 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 = request.user
                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(schema=schema, table=table)
            review_table.set_is_reviewed()
            metadata = self.load_json(schema, table, review_id=review_id)

            recursive_update(metadata, review_data)

            save_metadata_to_db(schema, table, metadata)

            if active_peer_review:
                # Update the oemetadata in the active PeerReview
                active_peer_review.oemetadata = metadata
                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)` #

Sorts the metadata of a table into categories and adds the value suggestion and comment that were added during the review, to facilitate Further processing easier.

Note

The categories spatial & temporal are often combined during visualization.

Parameters:

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

Returns:

Examples:

A return value can look like the below dictionary:

>>>
{
    "general": [
        {
        "field": "id",
        "value": "http: //127.0.0.1:8000/dataedit/view/model_draft/test2",
        "newValue": "",
        "reviewer_suggestion": "",
        "suggestion_comment": ""
        }
    ],
    "spatial": [...],
    "temporal": [...],
    "source": [...],
    "license": [...],
}

Source code in dataedit/views.py

def sort_in_category(self, schema, table, oemetadata):
    """
    Sorts the metadata of a table into categories and adds the value
    suggestion and comment that were added during the review, to facilitate
    Further processing easier.

    Note:
        The categories spatial & temporal are often combined during visualization.

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

    Returns:


    Examples:
        A return value can look like the below dictionary:

        >>>
        {
            "general": [
                {
                "field": "id",
                "value": "http: //127.0.0.1:8000/dataedit/view/model_draft/test2",
                "newValue": "",
                "reviewer_suggestion": "",
                "suggestion_comment": ""
                }
            ],
            "spatial": [...],
            "temporal": [...],
            "source": [...],
            "license": [...],
        }

    """

    val = self.parse_keys(oemetadata)
    gen_key_list = []
    spatial_key_list = []
    temporal_key_list = []
    source_key_list = []
    license_key_list = []

    for i in val:
        fieldKey = list(i.values())[0]
        if fieldKey.split(".")[0] == "spatial":
            spatial_key_list.append(i)
        elif fieldKey.split(".")[0] == "temporal":
            temporal_key_list.append(i)
        elif fieldKey.split(".")[0] == "sources":
            source_key_list.append(i)
        elif fieldKey.split(".")[0] == "licenses":
            license_key_list.append(i)

        elif (
            fieldKey.split(".")[0] == "name"
            or fieldKey.split(".")[0] == "title"
            or fieldKey.split(".")[0] == "id"
            or fieldKey.split(".")[0] == "description"
            or fieldKey.split(".")[0] == "language"
            or fieldKey.split(".")[0] == "subject"
            or fieldKey.split(".")[0] == "keywords"
            or fieldKey.split(".")[0] == "publicationDate"
            or fieldKey.split(".")[0] == "context"
        ):
            gen_key_list.append(i)

    meta = {
        "general": gen_key_list,
        "spatial": spatial_key_list,
        "temporal": temporal_key_list,
        "source": source_key_list,
        "license": license_key_list,
    }

    return meta

PeerRreviewContributorView#

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

Bases: PeerReviewView

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 PeerRreviewContributorView(PeerReviewView):
    """
    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, schema, table, review_id):
        """
        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.schema, peer_review.table)
        if not request.user.is_anonymous:
            level = request.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.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, schema, table, review_id):
        """
        Handle POST requests for contributor's review. Merges and updates
        the review data in the PeerReview table.

        Missing parts:
            - merge contributor field review and reviewer field review

        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.

        Note:
            This method has some missing parts regarding the merging of contributor
            and reviewer field 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")
            # unused
            # review_state = review_data.get("reviewFinished")
            current_opr = PeerReviewManager.filter_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`		Rendered HTML response for contributor review.

Source code in dataedit/views.py

def get(self, request, schema, table, review_id):
    """
    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.schema, peer_review.table)
    if not request.user.is_anonymous:
        level = request.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.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.

Missing parts

merge contributor field review and reviewer field review

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`		Rendered HTML response for contributor review.

Note

This method has some missing parts regarding the merging of contributor and reviewer field review.

Source code in dataedit/views.py

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

    Missing parts:
        - merge contributor field review and reviewer field review

    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.

    Note:
        This method has some missing parts regarding the merging of contributor
        and reviewer field 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")
        # unused
        # review_state = review_data.get("reviewFinished")
        current_opr = PeerReviewManager.filter_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.

`get_readable_table_name(table_obj)` #

get readable table name from metadata

Parameters:

Name	Type	Description	Default
`table_obj`	`object`	django orm	required

Returns:

Type	Description
`str`	str

Source code in dataedit/helper.py

def get_readable_table_name(table_obj: Table) -> str:
    """get readable table name from metadata

    Args:
        table_obj (object): django orm

    Returns:
        str
    """

    try:
        label = read_label(table_obj.name, table_obj.oemetadata)
    except Exception as e:
        raise e
    return label

`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

`process_review_data(review_data, metadata, categories)` #

Process the review data and update the metadata with the latest reviews and suggestions.

Parameters:

Name	Type	Description	Default
`review_data`	`list`	A list of dictionaries containing review data for each field.	required
`metadata`	`dict`	The original metadata object that needs to be updated.	required
`categories`	`list`	A list of categories in the metadata.	required

Returns:

Name	Type	Description
`dict`		A state dictionary containing the state of each field after processing the review data.

Note

The function sorts the fieldReview entries by timestamp (newest first) and updates the metadata with the latest reviewer suggestions, comments, and new values. The resulting state dictionary indicates the state of each field after processing.

Source code in dataedit/helper.py

def process_review_data(review_data, metadata, categories):
    """
    Process the review data and update the metadata with the latest reviews
    and suggestions.

    Args:
        review_data (list): A list of dictionaries containing review data for
                            each field.
        metadata (dict): The original metadata object that needs to be updated.
        categories (list): A list of categories in the metadata.

    Returns:
        dict: A state dictionary containing the state of each field
            after processing the review data.

    Note:
        The function sorts the fieldReview entries by timestamp (newest first)
        and updates the metadata with the latest reviewer suggestions,
        comments, and new values. The resulting state dictionary indicates
        the state of each field after processing.
    """
    state_dict = {}

    for review in review_data:
        field_key = review.get("key")
        field_review = review.get("fieldReview")

        if isinstance(field_review, list):
            # Sortiere die fieldReview-Einträge nach dem timestamp (neueste zuerst)
            sorted_field_review = sorted(
                field_review, key=lambda x: x.get("timestamp"), reverse=True
            )
            latest_field_review = (
                sorted_field_review[0] if sorted_field_review else None
            )

            if latest_field_review:
                state = latest_field_review.get("state")
                reviewer_suggestion = latest_field_review.get("reviewerSuggestion")
                reviewer_suggestion_comment = latest_field_review.get("comment")
                newValue = latest_field_review.get("newValue")
            else:
                state = None
                reviewer_suggestion = None
                reviewer_suggestion_comment = None
                newValue = None
        else:
            state = field_review.get("state")
            reviewer_suggestion = field_review.get("reviewerSuggestion")
            reviewer_suggestion_comment = field_review.get("comment")
            newValue = field_review.get("newValue")

        if reviewer_suggestion is not None and reviewer_suggestion_comment is not None:
            for category in categories:
                for item in metadata[category]:
                    if item["field"] == field_key:
                        item["reviewer_suggestion"] = reviewer_suggestion
                        item["suggestion_comment"] = reviewer_suggestion_comment
                        break

        if newValue is not None:
            for category in categories:
                for item in metadata[category]:
                    if item["field"] == field_key:
                        item["newValue"] = newValue
                        break

        state_dict[field_key] = state

    return state_dict

`read_label(table, oemetadata)` #

Extracts the readable name from @comment and appends the real name in parens. If comment is not a JSON-dictionary or does not contain a field 'Name' None is returned.

:param table: Name to append

:param comment: String containing a JSON-dictionary according to @Metadata

:return: Readable name appended by the true table name as string or None

Source code in dataedit/helper.py

def read_label(table, oemetadata) -> str:
    """
    Extracts the readable name from @comment and appends the real name in parens.
    If comment is not a JSON-dictionary or does not contain a field 'Name' None
    is returned.

    :param table: Name to append

    :param comment: String containing a JSON-dictionary according to @Metadata

    :return: Readable name appended by the true table name as string or None
    """
    try:
        if oemetadata.get("title"):
            return oemetadata["title"].strip() + " (" + table + ")"
        elif oemetadata.get("Title"):
            return oemetadata["Title"].strip() + " (" + table + ")"

        else:
            return None

    except Exception:
        return None

`recursive_update(metadata, review_data)` #

Recursively update the metadata with new values from the review data.

Parameters:

Name	Type	Description	Default
`metadata`	`dict`	The original metadata dictionary to be updated.	required
`review_data`	`dict`	The review data containing new values for various keys.	required

Note

The function traverses the review data, and for each key, it updates the corresponding value in the metadata if a new value is present and is not an empty string.

Source code in dataedit/helper.py

def recursive_update(metadata, review_data):
    """
    Recursively update the metadata with new values from the review data.

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

    Note:
        The function traverses the review data, and for each key, it updates the
        corresponding value in the metadata if a new value is present and is not
        an empty string.
    """

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

        if isinstance(review_key["fieldReview"], list):
            for field_review in review_key["fieldReview"]:
                new_value = field_review.get("newValue", None)
                if new_value is not None and new_value != "":
                    set_nested_value(metadata, keys, new_value)
        else:
            new_value = review_key["fieldReview"].get("newValue", None)
            if new_value is not None and new_value != "":
                set_nested_value(metadata, keys, new_value)

`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

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, table, 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.
    """

    from dataedit.models import Table

    # Load the table object
    table_obj = Table.load(schema=schema, table=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
`schema`	`str`	The name of the OEP schema.	required
`table`	`str`	The name of the table in the OEP schema.	required

Returns:

Name	Type	Description
`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(schema, table):
    """
    Load metadata for a specific table from the OEP database.

    Args:
        schema (str): The name of the OEP schema.
        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).
    """

    from dataedit.models import Table

    metadata = Table.load(schema=schema, table=table).oemetadata

    metadata = parse_meta_data(metadata, schema, table)
    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.
`schema`	`CharField`	Name of the schema where the table is located.
`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.
        schema (CharField): Name of the schema where the table is located.
        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)
    schema = 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, schema, table):
        """
        Load the current reviewer user.
        The current review is review is determened by the latest date started.

        Args:
            schema (string): Schema name
            table (string): Table name

        Returns:
            opr (PeerReview): PeerReview object related to the latest
            date started.
        """
        opr = (
            PeerReview.objects.filter(table=table, schema=schema)
            .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, schema, table):
        """
        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 schema
        # and table name
        peer_reviews = PeerReview.objects.filter(table=table, schema=schema).order_by(
            "date_started"
        )

        current_index = None
        for index, review in enumerate(peer_reviews):
            if review.id == self.id:
                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
            # prev_review, next_review = self.get_prev_and_next_reviews(
            #   self.schema, self.table
            # )

            # print(prev_review.id, next_review)
            # print(prev_review, next_review)
            # Create a new PeerReviewManager entry for this PeerReview
            # pm_new = PeerReviewManager(opr=self, prev_review=prev_review)

            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(
                    schema=self.schema, 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(
                    schema=self.schema, 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 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, schema, *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
            schema (str): Table database schema aka data topic

        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(schema=schema, table=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, to_schema, **kwargs
    ):
        # all_peer_reviews = self.objects.filter(table=table, schema=from_schema)
        # for peer_review in all_peer_reviews:
        if isinstance(self.review, str):
            review_data = json.loads(self.review)
        else:
            review_data = self.review

        review_data["topic"] = to_schema

        self.review = review_data
        self.schema = to_schema

        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

`get_prev_and_next_reviews(schema, 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, schema, table):
    """
    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 schema
    # and table name
    peer_reviews = PeerReview.objects.filter(table=table, schema=schema).order_by(
        "date_started"
    )

    current_index = None
    for index, review in enumerate(peer_reviews):
        if review.id == self.id:
            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(schema, table)` `classmethod` #

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

Parameters:

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

Returns:

Name	Type	Description
`opr`	`PeerReview`	PeerReview object related to the latest
		date started.

Source code in dataedit/models.py

@classmethod
def load(cls, schema, table):
    """
    Load the current reviewer user.
    The current review is review is determened by the latest date started.

    Args:
        schema (string): Schema name
        table (string): Table name

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

`set_version_of_metadata_for_review(table, schema, *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
`schema`	`str`	Table database schema aka data topic	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, schema, *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
        schema (str): Table database schema aka data topic

    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(schema=schema, table=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)
        # print(self.is_open_since, self.status)
        # 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(schema, table):
        """
        Get the contributor for the table a review is started.

        Args:
            schema (str): Schema name.
            table (str): Table name.

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

    @staticmethod
    def load_reviewer(schema, table):
        """
            Get the reviewer for the table a review is started.
        .
            Args:
                schema (str): Schema name.
                table (str): Table name.

            Returns:
                User: The reviewer user.
        """
        current_review = PeerReview.load(schema=schema, 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(schema, table):
        """
        Filter peer reviews by schema and table.

        Args:
            schema (str): Schema name.
            table (str): Table name.

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

    def filter_opr_by_id(opr_id):
        return PeerReview.objects.filter(id=opr_id).first()

`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(schema, table)` `staticmethod` #

Filter peer reviews by schema and table.

Parameters:

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

Returns:

Name	Type	Description
`QuerySet`		Filtered peer reviews.

Source code in dataedit/models.py

@staticmethod
def filter_opr_by_table(schema, table):
    """
    Filter peer reviews by schema and table.

    Args:
        schema (str): Schema name.
        table (str): Table name.

    Returns:
        QuerySet: Filtered peer reviews.
    """
    return PeerReview.objects.filter(schema=schema, 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(schema, table)` `staticmethod` #

Get the contributor for the table a review is started.

Parameters:

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

Returns:

Name	Type	Description
`User`		The contributor user.

Source code in dataedit/models.py

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

    Args:
        schema (str): Schema name.
        table (str): Table name.

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

`load_reviewer(schema, table)` `staticmethod` #

Get the reviewer for the table a review is started.

. Args: schema (str): Schema name. table (str): Table name.

Returns:
    User: The reviewer user.

Source code in dataedit/models.py

@staticmethod
def load_reviewer(schema, table):
    """
        Get the reviewer for the table a review is started.
    .
        Args:
            schema (str): Schema name.
            table (str): Table name.

        Returns:
            User: The reviewer user.
    """
    current_review = PeerReview.load(schema=schema, 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)
    # print(self.is_open_since, self.status)
    # 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#

get_readable_table_name(table_obj) #

get_review_for_key(key, review_data) #

merge_field_reviews(current_json, new_json) #

process_review_data(review_data, metadata, categories) #

read_label(table, oemetadata) #

recursive_update(metadata, review_data) #

set_nested_value(metadata, keys, value) #

Metadata Functions#

Save Metadata to Database#

Load Metadata from Database#

Models#

PeerReview#

get_prev_and_next_reviews(schema, table) #

load(schema, table) classmethod #

set_version_of_metadata_for_review(table, schema, *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(schema, table) staticmethod #

load(opr) classmethod #

load_contributor(schema, table) staticmethod #

load_reviewer(schema, 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)` #

`get_readable_table_name(table_obj)` #

`get_review_for_key(key, review_data)` #

`merge_field_reviews(current_json, new_json)` #

`process_review_data(review_data, metadata, categories)` #

`read_label(table, oemetadata)` #

`recursive_update(metadata, review_data)` #

`set_nested_value(metadata, keys, value)` #

`get_prev_and_next_reviews(schema, table)` #

`load(schema, table)` `classmethod` #

`set_version_of_metadata_for_review(table, schema, *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(schema, table)` `staticmethod` #

`load(opr)` `classmethod` #

`load_contributor(schema, table)` `staticmethod` #

`load_reviewer(schema, table)` `staticmethod` #

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

`set_next_reviewer()` #

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

`whos_turn()` #