Function materialize_output_recursive has a Cognitive Complexity of 33 (exceeds 5 allowed). Consider refactoring.

Open

    def materialize_output_recursive(self,
                                     operator_info: dict,
                                     output_field: dict,
                                     used_objects: list[str],
                                     objects_bucket: ObjectsBucket,

• Read up
  Read up

Function create_object_mutation_edges has a Cognitive Complexity of 29 (exceeds 5 allowed). Consider refactoring.

Open

    def create_object_mutation_edges(self, object_nodes: dict, mutation_nodes: dict):
        """Updates the dependency graph with edges between objects and mutations. 3 cases:
           Case 1: M -> O | When object(O) depends on mutation(M), means O has M in its "associatedMutations", weight 100
           Case 2: O -> M | When mutation(M) depends on object(O), means M has O in its "hardDependsOn", weight 100
           Case 3: O -> M | When mutation(M) depends on object(O), means M has O in its "softDependsOn", weight 1

• Read up
  Read up

Function __perform_dfs has a Cognitive Complexity of 24 (exceeds 5 allowed). Consider refactoring.

Open

    def __perform_dfs(self, starter_stack: list[Node], filter_mutation_type: list[str]):
        """Performs DFS with the initial starter stack

        Args:
            starter_stack (list[Node]): A list of the nodes to start the fuzzing

• Read up
  Read up

Function create_object_query_edges has a Cognitive Complexity of 21 (exceeds 5 allowed). Consider refactoring.

Open

    def create_object_query_edges(self, object_nodes: dict, query_nodes: dict):
        """Updates the dependency graph with edges in between objects and queries. 3 cases:
           Case 1: M -> O | When object(O) is produced by query(Q), means O has Q in its "associatedQueries", weight 100
           Case 2: O -> Q | When query(Q) depends on object(O), means Q has O in its "hardDependsOn", weight 100
           Case 3: O -> Q | When query(Q) depends on object(O), means Q has O in its "softDependsOn", weight 1

• Read up
  Read up

Function __run_mutation has a Cognitive Complexity of 21 (exceeds 5 allowed). Consider refactoring.

Open

    def __run_mutation(self, mutation_name: str, objects_bucket: ObjectsBucket, materializer: Materializer) -> tuple[dict, Result]:
        """Runs the mutation, and returns a new objects bucket. Performs a few things:
           1. Materializes the mutation with its parameters (resolving any dependencies from the object_bucket)
           2. Send the mutation against the server and gets the parses the object from the response
           3. Process the result in the objects_bucket if it's an object with an ID

• Read up
  Read up

Function materialize_input_recursive has a Cognitive Complexity of 20 (exceeds 5 allowed). Consider refactoring.

Open

    def materialize_input_recursive(self,
                                    operator_info: dict,
                                    input_field: dict,
                                    objects_bucket: ObjectsBucket,
                                    input_name: str,

• Read up
  Read up

Function get_random_scalar has a Cognitive Complexity of 18 (exceeds 5 allowed). Consider refactoring.

Open

    def get_random_scalar(self, input_name: str, scalar_type: str, objects_bucket: ObjectsBucket) -> str:
        """Gets a random scalar based on the scalar type, the return value will
           be a string regardless if of it's type as this function meant to be used
           during materialization
        Args:

• Read up
  Read up

File fuzzer.py has 260 lines of code (exceeds 250 allowed). Consider refactoring.

Open

"""Class for fuzzer

1. Gets all nodes that can be run without a dependency (query/mutation)
2. Adds these to the DFS queue
3. 1st Pass: Perform DFS, going through only creation nodes

File materializer.py has 258 lines of code (exceeds 250 allowed). Consider refactoring.

Open

"""Materializer:
Base class for a regular materializer
"""

from ..exceptions.hard_dependency_not_met_exception import HardDependencyNotMetException

ObjectsBucket has 21 functions (exceeds 20 allowed). Consider refactoring.

Open

class ObjectsBucket:
    def __init__(self, api: API):
        self.api = api

        # Stores {object_name: {type: str, results: dict}} where list is a result with the scalar fields of the object

Function retry has a Cognitive Complexity of 16 (exceeds 5 allowed). Consider refactoring.

Open

    def retry(self, url: str, payload: str | dict | list, gql_response: dict, retry_count) -> tuple[dict, bool]:
        """Retries the payload based on the error

        Args:
            url (str): The url of the endpoint

• Read up
  Read up

Function parse_gql_object has a Cognitive Complexity of 15 (exceeds 5 allowed). Consider refactoring.

Open

    def parse_gql_object(self, gql_object: dict) -> dict:
        """Parse a single object, noting down all of the other object this object depends on
           hardDependsOn: Where the object has fields on other objects and it's kind is NON_NULL
           softDependsOn: Where the object has fields on other objects and it can be NULL

• Read up
  Read up

Function resolve_inputs_related_to_ids_to_objects has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

    def resolve_inputs_related_to_ids_to_objects(self, endpoint_name: str, inputs_related_to_ids: dict, objects: dict) -> dict:
        """Resolves inputs related to IDs by looking at the name of the parameter after the ID string is removed

        Args:
            endpoint_name (str): The name of the query or mutation for these inputs

• Read up
  Read up

Function __run_query has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

    def __run_query(self, query_name: str, objects_bucket: ObjectsBucket, materializer: Materializer) -> tuple[dict, Result]:
        """Runs the query, and returns a new objects bucket

        Args:
            query_name (str): The name of the query

• Read up
  Read up

Function __init__ has 34 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    def __init__(self, save_path: str, url: str):
        """Initializes the compiler,
            creates all necessary file paths to save the outputs for run if doesn't already exist

        Args:

Function materialize_output_recursive has 7 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def materialize_output_recursive(self,

Function materialize_input_recursive has 7 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def materialize_input_recursive(self,

Function __evaluate has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def __evaluate(self, node: Node, visit_path: list[Node], check_hard_depends_on: bool = True) -> tuple[list[list[Node]], Result]:
        """Evaluates the path, performing the following based on the type of node:
           Case 1: If it's an object node, then we should check if the object is in our bucket. If not, fail, if it is,
                   then queue up the next neighboring nodes to visit
           Case 2: If it's an query node or mutation node, run the payload with the required objects, then store the results in the object bucket

• Read up
  Read up

Function find_key_in_dict has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def find_key_in_dict(self, dictionary: dict, key: str) -> tuple[str, str | int | float | bool | None]:
        """Recursively searches for the key in a nested dictionary and returns its full path and value.

        Args:
            dictionary (dict): The dictionary to search

• Read up
  Read up

Function check_is_data_empty has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def check_is_data_empty(result: dict) -> bool:
    for value in result.values():
        if isinstance(value, dict):
            if not check_is_data_empty(value):
                return False

• Read up
  Read up