Method get_command_policies

redis/_parsers/commands.py:552–692

Retrieve and process the command policies for all commands and subcommands. This method traverses through commands and subcommands, extracting policy details from associated data structures and constructing a dictionary of commands with their associated policies. It

(self)

            raise ValueError(f"Command {command_name} not found in commands")

    async def get_command_policies(self) -> Awaitable[PolicyRecords]:
        """
        Retrieve and process the command policies for all commands and subcommands.

        This method traverses through commands and subcommands, extracting policy details
        from associated data structures and constructing a dictionary of commands with their
        associated policies. It supports nested data structures and handles both main commands
        and their subcommands.

        Returns:
            PolicyRecords: A collection of commands and subcommands associated with their
            respective policies.

        Raises:
            IncorrectPolicyType: If an invalid policy type is encountered during policy extraction.
        """
        command_with_policies = {}

        def extract_policies(data, module_name, command_name):
            """
            Recursively extract policies from nested data structures.

            Args:
                data: The data structure to search (can be list, dict, str, bytes, etc.)
                command_name: The command name to associate with found policies
            """
            if isinstance(data, (str, bytes)):
                # Decode bytes to string if needed
                policy = str_if_bytes(data.decode())

                # Check if this is a policy string
                if policy.startswith("request_policy") or policy.startswith(
                    "response_policy"
                ):
                    if policy.startswith("request_policy"):
                        policy_type = policy.split(":")[1]

                        try:
                            command_with_policies[module_name][
                                command_name
                            ].request_policy = RequestPolicy(policy_type)
                        except ValueError:
                            raise IncorrectPolicyType(
                                f"Incorrect request policy type: {policy_type}"
                            )

                    if policy.startswith("response_policy"):
                        policy_type = policy.split(":")[1]

                        try:
                            command_with_policies[module_name][
                                command_name
                            ].response_policy = ResponsePolicy(policy_type)
                        except ValueError:
                            raise IncorrectPolicyType(
                                f"Incorrect response policy type: {policy_type}"
                            )