Technical docs

Views

Django views.

PeerReviewView

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

Bases: LoginRequiredMixin, View

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

Source code in dataedit/views.py

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

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

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

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

        return metadata

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

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

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

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

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

        Returns:
            HttpResponse: Rendered HTML response.
        """

        table_obj = table_or_404(table=table)
        topic = list(table_obj.topics.values_list("name", flat=True))

        # review_state = PeerReview.is_finished  # TODO: Use later
        json_schema = self.load_json_schema()
        can_add = False
        field_descriptions = get_all_field_descriptions(json_schema)

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

        oemetadata = self.load_json(table, review_id)
        metadata = sort_in_category(
            oemetadata=oemetadata
        )  # Generate URL for peer_review_reviewer
        if review_id is not None:
            url_peer_review = reverse(
                "dataedit:peer_review_reviewer",
                kwargs={"table": table, "review_id": review_id},
            )
            opr_review = PeerReviewManager.get_opr_by_id(opr_id=review_id)

            existing_review = (opr_review.review or {}).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
            )
            field_history_data = field_history(existing_review)
            # Read-only unless it is the reviewer's turn (and not finished).
            manager = PeerReviewManager.objects.filter(opr=opr_review).first()
            read_only = bool(review_finished) or (
                manager is not None and manager.current_reviewer != "reviewer"
            )
        else:
            url_peer_review = reverse(
                "dataedit:peer_review_create",
                kwargs={"table": table},
            )
            # existing_review={}
            state_dict = None
            review_finished = None
            field_history_data = {}
            read_only = False

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

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

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

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

        Args:
            request (HttpRequest): The incoming HTTP POST request.
            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 topic (TODO).
        """
        table_obj = table_or_404(table=table)

        # get the review data and additional application metadata
        # from user peer review submit/save
        review_data = json.loads(request.body)

        # The delete path is handled directly (unified into ReviewService in a
        # later step); everything else is orchestrated by the service.
        if review_data.get("reviewType") == "delete":
            return delete_peer_review(review_id)

        service = ReviewService(table_name=table_obj.name, actor=request.user)
        try:
            service.submit_reviewer_review(review_data, review_id=review_id)
        except ContributorNotFoundError as exc:
            return JsonResponse({"error": str(exc)}, status=400)
        except (ReviewFinishedError, NotYourTurnError) as exc:
            return JsonResponse({"error": str(exc)}, status=409)

        return JsonResponse({"status": "success"}, status=200)

get(request, 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
table	str	The name of the table.	required
review_id	int	The ID of the review. Defaults to None.	None

Returns:

Name	Type	Description
HttpResponse	HttpResponse	Rendered HTML response.

Source code in dataedit/views.py

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

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

    Returns:
        HttpResponse: Rendered HTML response.
    """

    table_obj = table_or_404(table=table)
    topic = list(table_obj.topics.values_list("name", flat=True))

    # review_state = PeerReview.is_finished  # TODO: Use later
    json_schema = self.load_json_schema()
    can_add = False
    field_descriptions = get_all_field_descriptions(json_schema)

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

    oemetadata = self.load_json(table, review_id)
    metadata = sort_in_category(
        oemetadata=oemetadata
    )  # Generate URL for peer_review_reviewer
    if review_id is not None:
        url_peer_review = reverse(
            "dataedit:peer_review_reviewer",
            kwargs={"table": table, "review_id": review_id},
        )
        opr_review = PeerReviewManager.get_opr_by_id(opr_id=review_id)

        existing_review = (opr_review.review or {}).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
        )
        field_history_data = field_history(existing_review)
        # Read-only unless it is the reviewer's turn (and not finished).
        manager = PeerReviewManager.objects.filter(opr=opr_review).first()
        read_only = bool(review_finished) or (
            manager is not None and manager.current_reviewer != "reviewer"
        )
    else:
        url_peer_review = reverse(
            "dataedit:peer_review_create",
            kwargs={"table": table},
        )
        # existing_review={}
        state_dict = None
        review_finished = None
        field_history_data = {}
        read_only = False

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

load_json(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
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, table: str, review_id=None):
    """
    Load JSON metadata from the database. If the review_id is available
    then load the metadata form the peer review instance and not from the
    table. This avoids changes to the metadata that is or was reviewed.

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

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

    return metadata

load_json_schema()

Load the JSON schema used for validating metadata.

Note

Update this method if a new oemetadata version is released.

Returns:

Name	Type	Description
dict		JSON schema.

Source code in dataedit/views.py

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

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

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

post(request, 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
table	str	The name of the table.	required
review_id	int	The ID of the review. Defaults to None.	None

Returns:

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

Raises:

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

Note

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

Source code in dataedit/views.py

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

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

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

    Args:
        request (HttpRequest): The incoming HTTP POST request.
        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 topic (TODO).
    """
    table_obj = table_or_404(table=table)

    # get the review data and additional application metadata
    # from user peer review submit/save
    review_data = json.loads(request.body)

    # The delete path is handled directly (unified into ReviewService in a
    # later step); everything else is orchestrated by the service.
    if review_data.get("reviewType") == "delete":
        return delete_peer_review(review_id)

    service = ReviewService(table_name=table_obj.name, actor=request.user)
    try:
        service.submit_reviewer_review(review_data, review_id=review_id)
    except ContributorNotFoundError as exc:
        return JsonResponse({"error": str(exc)}, status=400)
    except (ReviewFinishedError, NotYourTurnError) as exc:
        return JsonResponse({"error": str(exc)}, status=409)

    return JsonResponse({"status": "success"}, status=200)

PeerReviewContributorView

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

Bases: TablePeerReviewView

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

Source code in dataedit/views.py

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

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

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

        Returns:
            HttpResponse: Rendered HTML response for contributor review.
        """
        table_obj = table_or_404(table=table)

        can_add = False
        peer_review = PeerReview.objects.get(id=review_id)

        user: login_models.myuser = request.user  # type: ignore
        if not user.is_anonymous:
            level = user.get_table_permission_level(table_obj)
            can_add = level >= login.permissions.WRITE_PERM
        oemetadata = self.load_json(table_obj.name, review_id)
        metadata = sort_in_category(oemetadata=oemetadata)
        json_schema = self.load_json_schema()
        field_descriptions = get_all_field_descriptions(json_schema)
        review_data = (peer_review.review or {}).get("reviews", [])

        categories = [
            "general",
            "spatial",
            "temporal",
            "source",
            "license",
        ]
        state_dict = process_review_data(
            review_data=review_data, metadata=metadata, categories=categories
        )
        review_finished = peer_review.is_finished
        # Read-only unless it is the contributor's turn (and not finished).
        manager = PeerReviewManager.objects.filter(opr=peer_review).first()
        read_only = bool(review_finished) or (
            manager is not None and manager.current_reviewer != "contributor"
        )
        context_meta = {
            "config": json.dumps(
                {
                    "can_add": can_add,
                    "url_peer_review": reverse(
                        "dataedit:peer_review_contributor",
                        kwargs={
                            "table": table_obj.name,
                            "review_id": review_id,
                        },
                    ),
                    "url_table": reverse(
                        "dataedit:view", kwargs={"table": table_obj.name}
                    ),
                    # "topic": table_obj.topics,
                    "table": table_obj.name,
                    "review_finished": review_finished,
                    "read_only": read_only,
                    "review_id": review_id,
                }
            ),
            "table": table_obj.name,
            # "topic": table_obj.topics,
            "meta": metadata,
            "json_schema": json_schema,
            "field_descriptions_json": json.dumps(field_descriptions),
            "state_dict": json.dumps(state_dict),
            "field_history_json": json.dumps(field_history(review_data)),
            "review_finished": review_finished,
        }
        return render(request, "dataedit/opr_contributor.html", context=context_meta)

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

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

        Returns:
            HttpResponse: Rendered HTML response for contributor review.

        """
        review_data = json.loads(request.body)
        service = ReviewService(table_name=table, actor=request.user)
        try:
            service.submit_contributor_review(review_data, review_id=review_id)
        except (ReviewFinishedError, NotYourTurnError) as exc:
            return JsonResponse({"error": str(exc)}, status=409)
        return JsonResponse({"status": "success"}, status=200)

get(request, 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
table	str	The name of the table.	required
review_id	int	The ID of the review.	required

Returns:

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

Source code in dataedit/views.py

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

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

    Returns:
        HttpResponse: Rendered HTML response for contributor review.
    """
    table_obj = table_or_404(table=table)

    can_add = False
    peer_review = PeerReview.objects.get(id=review_id)

    user: login_models.myuser = request.user  # type: ignore
    if not user.is_anonymous:
        level = user.get_table_permission_level(table_obj)
        can_add = level >= login.permissions.WRITE_PERM
    oemetadata = self.load_json(table_obj.name, review_id)
    metadata = sort_in_category(oemetadata=oemetadata)
    json_schema = self.load_json_schema()
    field_descriptions = get_all_field_descriptions(json_schema)
    review_data = (peer_review.review or {}).get("reviews", [])

    categories = [
        "general",
        "spatial",
        "temporal",
        "source",
        "license",
    ]
    state_dict = process_review_data(
        review_data=review_data, metadata=metadata, categories=categories
    )
    review_finished = peer_review.is_finished
    # Read-only unless it is the contributor's turn (and not finished).
    manager = PeerReviewManager.objects.filter(opr=peer_review).first()
    read_only = bool(review_finished) or (
        manager is not None and manager.current_reviewer != "contributor"
    )
    context_meta = {
        "config": json.dumps(
            {
                "can_add": can_add,
                "url_peer_review": reverse(
                    "dataedit:peer_review_contributor",
                    kwargs={
                        "table": table_obj.name,
                        "review_id": review_id,
                    },
                ),
                "url_table": reverse(
                    "dataedit:view", kwargs={"table": table_obj.name}
                ),
                # "topic": table_obj.topics,
                "table": table_obj.name,
                "review_finished": review_finished,
                "read_only": read_only,
                "review_id": review_id,
            }
        ),
        "table": table_obj.name,
        # "topic": table_obj.topics,
        "meta": metadata,
        "json_schema": json_schema,
        "field_descriptions_json": json.dumps(field_descriptions),
        "state_dict": json.dumps(state_dict),
        "field_history_json": json.dumps(field_history(review_data)),
        "review_finished": review_finished,
    }
    return render(request, "dataedit/opr_contributor.html", context=context_meta)

post(request, table, review_id)

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

Parameters:

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

Returns:

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

Source code in dataedit/views.py

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

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

    Returns:
        HttpResponse: Rendered HTML response for contributor review.

    """
    review_data = json.loads(request.body)
    service = ReviewService(table_name=table, actor=request.user)
    try:
        service.submit_contributor_review(review_data, review_id=review_id)
    except (ReviewFinishedError, NotYourTurnError) as exc:
        return JsonResponse({"error": str(exc)}, status=409)
    return JsonResponse({"status": "success"}, status=200)

Helper Functions

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

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

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

add_tag(name, color)

Parameters:

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

Source code in dataedit/helper.py

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

change_requests(table_obj)

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

Source code in dataedit/helper.py

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

    display_message = None
    api_columns = get_column_changes(reviewed=False, table_obj=table_obj)
    api_constraints = get_constraints_changes(reviewed=False, table_obj=table_obj)

    data = dict()

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

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

    old_description = describe_columns(table_obj)

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

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

        old_cd = old_description.get(name)

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

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

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

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

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

        data["api_constraints"][id] = value

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

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

delete_peer_review(review_id)

Remove Peer Review by review_id.

Source code in dataedit/helper.py

def delete_peer_review(review_id):
    """Remove Peer Review by review_id."""
    if review_id:
        peer_review = PeerReview.objects.filter(id=review_id).first()
        if peer_review:
            peer_review.delete()
            return JsonResponse({"message": "PeerReview successfully deleted."})
        else:
            return JsonResponse({"error": "PeerReview not found."}, status=404)
    else:
        return JsonResponse({"error": "Review ID is required."}, status=400)

edit_tag(id, name, color)

Parameters:

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

Source code in dataedit/helper.py

def edit_tag(id: str, name: str, color: str) -> None:
    """
    Args:
        id(int): tag id
        name(str): max 40 character tag text
        color(str): hexadecimal color code, eg #aaf0f0

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

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

find tables given search criteria

Parameters:

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

Returns:

Type	Description
QuerySet[Table]	QuerySet of Table objetcs

Source code in dataedit/helper.py

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

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

    Returns:
        QuerySet of Table objetcs
    """

    tables = Table.objects

    tables = tables.filter(is_sandbox=False)

    if topic_name:
        if topic_name == PSEUDO_TOPIC_DRAFT:
            tables = tables.filter(is_publish=False)
        else:
            tables = tables.filter(topics__pk=topic_name)

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

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

    return tables

get_all_tags(table_name=None)

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

Source code in dataedit/helper.py

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

    return tags

get_review_for_key(key, review_data)

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

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

    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)

Attach reviewer fields (suggestions/comments/newValue) to metadata items.

The metadata[category] structures may be nested (dict/list) and can contain non-dict leaf values (e.g. strings). We therefore walk the structure recursively and only mutate leaf dicts that represent a field item.

Source code in dataedit/helper.py

def process_review_data(review_data, metadata, categories):
    """Attach reviewer fields (suggestions/comments/newValue) to metadata items.

    The `metadata[category]` structures may be nested (dict/list) and can contain
    non-dict leaf values (e.g. strings). We therefore walk the structure
    recursively and only mutate leaf dicts that represent a field item.
    """

    state_dict: dict[str, str | None] = {}

    def iter_field_items(node):
        """Yield all leaf field-item dicts inside nested list/dict structures.

        A leaf field item is a dict with a string key `field`.
        """
        if isinstance(node, list):
            for el in node:
                yield from iter_field_items(el)
        elif isinstance(node, dict):
            if isinstance(node.get("field"), str):
                yield node
            else:
                for v in node.values():
                    yield from iter_field_items(v)
        # Ignore other node types (e.g. str/int/None)

    # Initialize fields safely
    for category in categories:
        cat_node = metadata.get(category)
        if cat_node is None:
            continue
        for item in iter_field_items(cat_node):
            item.setdefault("reviewer_suggestion", "")
            item.setdefault("suggestion_comment", "")
            item.setdefault("additional_comment", "")
            item.setdefault("newValue", "")

    # Apply review values
    for review in review_data or []:
        field_key = review.get("key")
        field_review = review.get("fieldReview")
        category = review.get("category")

        state = None
        reviewer_suggestion = ""
        reviewer_suggestion_comment = ""
        newValue = ""
        additional_comment = ""

        if isinstance(field_review, list):
            sorted_field_review = sorted(
                field_review,
                key=lambda x: (x.get("timestamp") or 0) if isinstance(x, dict) else 0,
                reverse=True,
            )
            latest = sorted_field_review[0] if sorted_field_review else None
            if isinstance(latest, dict):
                state = latest.get("state")
                reviewer_suggestion = latest.get("reviewerSuggestion") or ""
                reviewer_suggestion_comment = latest.get("comment") or ""
                newValue = latest.get("newValue") or ""
                additional_comment = latest.get("additionalComment") or ""

        elif isinstance(field_review, dict):
            state = field_review.get("state")
            reviewer_suggestion = field_review.get("reviewerSuggestion") or ""
            reviewer_suggestion_comment = field_review.get("comment") or ""
            newValue = field_review.get("newValue") or ""
            additional_comment = field_review.get("additionalComment") or ""

        # Update the matching item in metadata for this category
        if category in metadata and field_key:
            for item in iter_field_items(metadata.get(category)):
                if item.get("field") == field_key:
                    item["reviewer_suggestion"] = reviewer_suggestion
                    item["suggestion_comment"] = reviewer_suggestion_comment
                    item["additional_comment"] = additional_comment
                    item["newValue"] = newValue
                    break

        if field_key:
            state_dict[field_key] = state

    return state_dict

recursive_update(metadata, review_data)

Recursively updates metadata with new values from review_data.

Skips or removes fields with status 'rejected'.

Source code in dataedit/helper.py

def recursive_update(metadata, review_data):
    """Recursively updates metadata with new values from review_data.

    Skips or removes fields with status 'rejected'.
    """

    def delete_nested_field(data: list | dict | None, keys: list[str]):
        """
        Removes a nested field from a dictionary based on a list of keys.

        Args:
            data (dict or list): The dictionary or list from which
            to remove the field.
            keys (list): A list of keys pointing to the field to remove.
        """

        for key in keys[:-1]:
            if isinstance(data, list):
                key = int(key)
                data = data[key]
            elif isinstance(data, dict):
                data = data.get(key)
            else:
                raise NotImplementedError()

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

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

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

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

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

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

    return metadata

set_nested_value(metadata, keys, value)

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

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

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

update_keywords_from_tags(table)

synchronize keywords in metadata with tags

Source code in dataedit/helper.py

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

    metadata = load_metadata_from_db(table=table.name)

    keywords = [tag.name_normalized for tag in table.tags.all()]
    metadata["resources"][0]["keywords"] = keywords

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

Metadata Functions

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

Save Metadata to Database

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

Parameters:

Name	Type	Description	Default
table	str	The name of the table in the OEP.	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(table: str, updated_metadata):
    """
    Save updated metadata for a specific table in the OEP database.

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

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

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

    # Update the oemetadata field
    table_obj.oemetadata = updated_metadata

    # Save the updated table object
    table_obj.save()

Load Metadata from Database

Load metadata for a specific table from the OEP database.

Parameters:

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

Returns:

Name	Type	Description
dict	dict	The loaded metadata dictionary.

Note

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

Source code in dataedit/metadata/__init__.py

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

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

    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).
    """

    table_obj = Table.load(name=table)
    metadata = table_obj.oemetadata
    if not metadata:
        # empty / new metadata

        # TODO: the template is full of empty strings, which are not valid metadata
        # so we use only parts of it

        metaMetadata = OEMETADATA_V20_TEMPLATE["metaMetadata"]
        name = table_obj.name

        metadata = {
            "name": name,
            "resources": [{"name": name}],
            "metaMetadata": metaMetadata,
        }
        validate_metadata(metadata, check_license=False)

    return metadata

Models

Django models.

PeerReview

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

Note

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

Bases: Model

Represents a peer review in the database.

Attributes:

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

Source code in dataedit/models.py

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

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

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

    review_id: QuerySet["PeerReviewManager"]  # related_name, for static type checking
    rounds: QuerySet["ReviewRound"]  # related_name, for static type checking

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

        Args:
            table (string): Table name

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

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

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

        prev_review = None
        next_review = None

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

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

        return prev_review, next_review

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

        if self.contributor == self.reviewer:
            raise ValidationError("Contributor and reviewer cannot be the same.")

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

        if review_type is None:
            return

        # One PeerReviewManager per review: get_or_create avoids the historical
        # duplicate-manager bug where every save() created a fresh manager
        # (which then broke PeerReviewManager.load via .get()).
        pm, _ = PeerReviewManager.objects.get_or_create(opr=self)

        if review_type == "save":
            pm.status = ReviewDataStatus.SAVED.value
            pm.save()

        elif review_type == "submit":
            result = self.set_version_of_metadata_for_review(table=self.table)
            logger.info(result[1])

            pm.status = ReviewDataStatus.SUBMITTED.value
            pm.set_next_reviewer()  # persists the manager

        elif review_type == "finished":
            result = self.set_version_of_metadata_for_review(table=self.table)
            logger.info(result[1])

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

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

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

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

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

        """

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

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

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

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

        Args:
            table (str): Table name

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

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

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

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

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

        review_data["topic"] = topic

        self.review = review_data

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

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

delete(*args, **kwargs)

Custom delete method to remove related PeerReviewManager entries.

Source code in dataedit/models.py

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

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

get_prev_and_next_reviews(table)

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

Source code in dataedit/models.py

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

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

    prev_review = None
    next_review = None

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

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

    return prev_review, next_review

load(table) classmethod

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

Parameters:

Name	Type	Description	Default
table	string	Table name	required

Returns:

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

Source code in dataedit/models.py

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

    Args:
        table (string): Table name

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

set_version_of_metadata_for_review(table, *args, **kwargs)

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

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

Parameters:

Name	Type	Description	Default
table	str	Table name	required

Returns:

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

Source code in dataedit/models.py

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

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

    Args:
        table (str): Table name

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

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

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

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

update(*args, **kwargs)

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

Source code in dataedit/models.py

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

    """

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

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

PeerReviewManager

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

Note

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

Bases: Model

Manages peer review processes.

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

Attributes:

Name	Type	Description
opr	ForeignKey	The associated peer review.
current_reviewer	CharField	The current reviewer.
status	CharField	The current status of the review.
is_open_since	CharField	How long the review has been open.

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

    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)

    # Review history (the ping-pong of turns) is modelled by ReviewRound, not by
    # prev/next FKs on the manager — those were unused and have been removed.

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

        Args:
            opr (PeerReview): The peer review.

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

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

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

    @classmethod
    def update_open_since(cls, opr: PeerReview | None = 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_manager = PeerReviewManager.objects.get(opr=opr)
        else:
            peer_review_manager = cls()

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

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

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

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

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

        return role, result

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

        Args:
            table (str): Table name.

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

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

        Args:
            table (str): Table name.

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

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

        Args:
            reviewer_user (User): The reviewer user.

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

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

        Args:
            contributor_user (User): The contributor user.

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

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

        Args:
            contributor_user (User): The contributor user.

        Returns:
            QuerySet: Filtered peer reviews.
        """

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

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

        Args:
            contributor_user (User): The contributor user.

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

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

        Args:
            table (str): Table name.

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

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

filter_latest_open_opr_by_contributor(contributor_user) staticmethod

Get the last open peer review for the given contributor.

Parameters:

Name	Type	Description	Default
contributor_user	User	The contributor user.	required

Returns:

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

Source code in dataedit/models.py

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

    Args:
        contributor_user (User): The contributor user.

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

filter_latest_open_opr_by_reviewer(reviewer_user) staticmethod

Get the last open peer review for the given contributor.

Parameters:

Name	Type	Description	Default
contributor_user	User	The contributor user.	required

Returns:

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

Source code in dataedit/models.py

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

    Args:
        contributor_user (User): The contributor user.

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

filter_opr_by_contributor(contributor_user) staticmethod

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

Parameters:

Name	Type	Description	Default
contributor_user	User	The contributor user.	required

Returns:

Name	Type	Description
QuerySet		Filtered peer reviews.

Source code in dataedit/models.py

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

    Args:
        contributor_user (User): The contributor user.

    Returns:
        QuerySet: Filtered peer reviews.
    """

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

filter_opr_by_reviewer(reviewer_user) staticmethod

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

Parameters:

Name	Type	Description	Default
reviewer_user	User	The reviewer user.	required

Returns:

Name	Type	Description
QuerySet		Filtered peer reviews.

Source code in dataedit/models.py

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

    Args:
        reviewer_user (User): The reviewer user.

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

filter_opr_by_table(table) staticmethod

Filter peer reviews by table.

Parameters:

Name	Type	Description	Default
table	str	Table name.	required

Returns:

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

Source code in dataedit/models.py

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

    Args:
        table (str): Table name.

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

load(opr) classmethod

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

Parameters:

Name	Type	Description	Default
opr	PeerReview	The peer review.	required

Returns:

Name	Type	Description
PeerReviewManager		The peer review manager.

Source code in dataedit/models.py

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

    Args:
        opr (PeerReview): The peer review.

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

load_contributor(table) staticmethod

Get the contributor for the table a review is started.

Parameters:

Name	Type	Description	Default
table	str	Table name.	required

Returns:

Name	Type	Description
User		The contributor user.

Source code in dataedit/models.py

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

    Args:
        table (str): Table name.

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

load_reviewer(table) staticmethod

Get the reviewer for the table a review is started.

Parameters:

Name	Type	Description	Default
table	str	Table name.	required

Returns:

Name	Type	Description
User		The reviewer user.

Source code in dataedit/models.py

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

    Args:
        table (str): Table name.

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

save(*args, **kwargs)

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

Source code in dataedit/models.py

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

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

set_next_reviewer()

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

Source code in dataedit/models.py

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

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

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

Parameters:

Name	Type	Description	Default
opr	PeerReview	The peer review.	None

Source code in dataedit/models.py

@classmethod
def update_open_since(cls, opr: PeerReview | None = 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_manager = PeerReviewManager.objects.get(opr=opr)
    else:
        peer_review_manager = cls()

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

    # Call the parent class's save method to save the PeerReviewManager instance
    peer_review_manager.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