Adding QoS functionalities

PyPowerStore/configuration.py

@@ -2631,3 +2631,115 @@ def _prepare_create_cluster_payload(is_http_redirect_enabled, **kwargs):
             payload["security_config"] = security_config
 
         return payload
+
+    # File IO Limit Rule methods (v4.1.0.0+)
+
+    def get_file_io_limit_rules(self, filter_dict=None, all_pages=False):
+        """Get a list of file IO limit rules.
+
+        :param filter_dict: (optional) Filter detail
+        :type filter_dict: dict
+        :param all_pages: (optional) Indicates whether to return all elements
+        :type all_pages: bool
+        :return: File IO limit rule list
+        :rtype: list of dict
+        """
+        LOG.info(
+            "Getting file_io_limit_rules with filter: '%s' and all_pages: %s",
+            filter_dict, all_pages,
+        )
+        querystring = helpers.prepare_querystring(
+            constants.SELECT_ID_AND_NAME, filter_dict,
+        )
+        return self.config_client.request(
+            constants.GET,
+            constants.FILE_IO_LIMIT_RULE_LIST_URL.format(self.server_ip),
+            payload=None,
+            querystring=querystring,
+            all_pages=all_pages,
+        )
+
+    def get_file_io_limit_rule_details(self, rule_id):
+        """Get details of a file IO limit rule.
+
+        :param rule_id: The file IO limit rule ID
+        :type rule_id: str
+        :return: File IO limit rule details
+        :rtype: dict
+        """
+        LOG.info("Getting file_io_limit_rule details by ID: '%s'", rule_id)
+        return self.config_client.request(
+            constants.GET,
+            constants.FILE_IO_LIMIT_RULE_OBJECT_URL.format(self.server_ip, rule_id),
+            payload=None,
+            querystring=constants.FILE_IO_LIMIT_RULE_DETAILS_QUERY,
+        )
+
+    def get_file_io_limit_rule_by_name(self, rule_name):
+        """Get file IO limit rule details by name.
+
+        :param rule_name: The file IO limit rule name
+        :type rule_name: str
+        :return: File IO limit rule details
+        :rtype: list of dict
+        """
+        LOG.info("Getting file_io_limit_rule details by name: '%s'", rule_name)
+        return self.config_client.request(
+            constants.GET,
+            constants.FILE_IO_LIMIT_RULE_LIST_URL.format(self.server_ip),
+            payload=None,
+            querystring=helpers.prepare_querystring(
+                constants.FILE_IO_LIMIT_RULE_DETAILS_QUERY,
+                name=constants.EQUALS + rule_name,
+            ),
+        )
+
+    def create_file_io_limit_rule(self, payload):
+        """Create a new file IO limit rule.
+
+        :param payload: Request payload (must contain 'name' and 'max_bw')
+        :type payload: dict
+        :return: ID of the created file IO limit rule
+        :rtype: dict
+        """
+        LOG.info("Creating file_io_limit_rule with payload: '%s'", payload)
+        return self.config_client.request(
+            constants.POST,
+            constants.FILE_IO_LIMIT_RULE_LIST_URL.format(self.server_ip),
+            payload=payload,
+        )
+
+    def modify_file_io_limit_rule(self, rule_id, payload):
+        """Modify a file IO limit rule.
+
+        :param rule_id: The file IO limit rule ID
+        :type rule_id: str
+        :param payload: Modification payload
+        :type payload: dict
+        :return: None
+        :rtype: None
+        """
+        LOG.info(
+            "Modifying file_io_limit_rule '%s' with payload: '%s'", rule_id, payload,
+        )
+        return self.config_client.request(
+            constants.PATCH,
+            constants.FILE_IO_LIMIT_RULE_OBJECT_URL.format(self.server_ip, rule_id),
+            payload=payload,
+        )
+
+    def delete_file_io_limit_rule(self, rule_id):
+        """Delete a file IO limit rule.
+
+        :param rule_id: The file IO limit rule ID
+        :type rule_id: str
+        :return: None
+        :rtype: None
+        """
+        LOG.info("Deleting file_io_limit_rule: '%s'", rule_id)
+        return self.config_client.request(
+            constants.DELETE,
+            constants.FILE_IO_LIMIT_RULE_OBJECT_URL.format(self.server_ip, rule_id),
+        )
+
+    # File IO Limit Rule methods end

PyPowerStore/protection.py

@@ -847,6 +847,104 @@ def delete_protection_policy(self, policy_id):
             constants.PROTECTION_POLICY_OBJECT_URL.format(self.server_ip, policy_id),
         )
 
+    # QoS / File_Performance Policy Methods
+
+    def get_policy_details(self, policy_id):
+        """Get details of a QoS or File_Performance policy by ID.
+
+        :param policy_id: Policy unique identifier.
+        :type policy_id: str
+        :return: Policy details.
+        :rtype: dict
+        """
+        LOG.info("Getting QoS policy details by ID: '%s'", policy_id)
+        # Use singular field names: io_limit_rule, file_io_limit_rule
+        # The API reference incorrectly shows plural forms
+        querystring = {
+            "select": "id,name,description,type,type_l10n,"
+            "io_limit_rule(id,name),file_io_limit_rule(id,name)",
+        }
+        return self.rest_client.request(
+            constants.GET,
+            constants.PROTECTION_POLICY_OBJECT_URL.format(self.server_ip, policy_id),
+            querystring=querystring,
+        )
+
+    def get_policy_by_name(self, name, policy_type=None):
+        """Get a QoS or File_Performance policy by name.
+
+        :param name: Policy name.
+        :type name: str
+        :param policy_type: Optional policy type filter (QoS or File_Performance).
+        :type policy_type: str
+        :return: List of matching policies.
+        :rtype: list[dict]
+        """
+        LOG.info("Getting QoS policy details by name: '%s'", name)
+        # Use singular field names: io_limit_rule, file_io_limit_rule
+        querystring = {
+            "select": "id,name,description,type,type_l10n,"
+            "io_limit_rule(id,name),file_io_limit_rule(id,name)",
+            "name": constants.EQUALS + name,
+        }
+        if policy_type:
+            querystring["type"] = constants.EQUALS + policy_type
+        return self.rest_client.request(
+            constants.GET,
+            constants.PROTECTION_POLICY_LIST_URL.format(self.server_ip),
+            querystring=querystring,
+        )
+
+    def create_policy(self, payload):
+        """Create a new QoS or File_Performance policy.
+
+        :param payload: Request payload containing 'name', and either
+                        'io_limit_rule_id' (QoS) or 'file_io_limit_rule_id'
+                        (File_Performance).
+        :type payload: dict
+        :return: Response dict with policy ID.
+        :rtype: dict
+        """
+        LOG.info("Creating QoS policy with payload: '%s'", payload)
+        return self.rest_client.request(
+            constants.POST,
+            constants.PROTECTION_POLICY_LIST_URL.format(self.server_ip),
+            payload,
+        )
+
+    def modify_policy(self, policy_id, payload):
+        """Modify a QoS or File_Performance policy.
+
+        :param policy_id: Policy unique identifier.
+        :type policy_id: str
+        :param payload: Modification payload.
+        :type payload: dict
+        :return: None
+        :rtype: None
+        """
+        LOG.info("Modifying QoS policy '%s' with payload: '%s'", policy_id, payload)
+        return self.rest_client.request(
+            constants.PATCH,
+            constants.PROTECTION_POLICY_OBJECT_URL.format(self.server_ip, policy_id),
+            payload,
+        )
+
+    def delete_policy(self, policy_id):
+        """Delete a QoS or File_Performance policy.
+
+        :param policy_id: Policy unique identifier.
+        :type policy_id: str
+        :return: None if success else raise exception.
+        :rtype: None
+        """
+        LOG.info("Deleting QoS policy: '%s'", policy_id)
+        return self.rest_client.request(
+            constants.DELETE,
+            constants.PROTECTION_POLICY_OBJECT_URL.format(self.server_ip, policy_id),
+        )
+
+    # QoS / File_Performance Policy Methods End
+
     # FS Snapshot Methods
 
     def get_filesystem_snapshot_details_by_name(