RDS - simplify filter-parameter logic (#4966)

moto/rds/responses.py

@@ -4,7 +4,6 @@ from moto.core.responses import BaseResponse
 from moto.ec2.models import ec2_backends
 from .models import rds_backends
 from .exceptions import DBParameterGroupNotFoundError
-from .utils import filters_from_querystring
 
 
 class RDSResponse(BaseResponse):
@@ -171,7 +170,8 @@ class RDSResponse(BaseResponse):
 
     def describe_db_instances(self):
         db_instance_identifier = self._get_param("DBInstanceIdentifier")
-        filters = filters_from_querystring(self.querystring)
+        filters = self._get_multi_param("Filters.Filter.")
+        filters = {f["Name"]: f["Values"] for f in filters}
         all_instances = list(
             self.backend.describe_databases(db_instance_identifier, filters=filters)
         )
@@ -240,7 +240,8 @@ class RDSResponse(BaseResponse):
     def describe_db_snapshots(self):
         db_instance_identifier = self._get_param("DBInstanceIdentifier")
         db_snapshot_identifier = self._get_param("DBSnapshotIdentifier")
-        filters = filters_from_querystring(self.querystring)
+        filters = self._get_multi_param("Filters.Filter.")
+        filters = {f["Name"]: f["Values"] for f in filters}
         snapshots = self.backend.describe_database_snapshots(
             db_instance_identifier, db_snapshot_identifier, filters
         )
@@ -549,7 +550,8 @@ class RDSResponse(BaseResponse):
     def describe_db_cluster_snapshots(self):
         db_cluster_identifier = self._get_param("DBClusterIdentifier")
         db_snapshot_identifier = self._get_param("DBClusterSnapshotIdentifier")
-        filters = filters_from_querystring(self.querystring)
+        filters = self._get_multi_param("Filters.Filter.")
+        filters = {f["Name"]: f["Values"] for f in filters}
         snapshots = self.backend.describe_db_cluster_snapshots(
             db_cluster_identifier, db_snapshot_identifier, filters
         )

moto/rds/utils.py

@@ -1,4 +1,3 @@
-import re
 from collections import namedtuple
 
 from botocore.utils import merge_dicts
@@ -18,35 +17,6 @@ FilterDef = namedtuple(
 )
 
 
-def filters_from_querystring(querystring):
-    """Parses filters out of the query string computed by the
-    moto.core.responses.BaseResponse class.
-
-    :param dict[str, list[str]] querystring:
-        The `moto`-processed URL query string dictionary.
-    :returns:
-        Dict mapping filter names to filter values.
-    :rtype:
-        dict[str, list[str]]
-    """
-    response_values = {}
-    for key, value in sorted(querystring.items()):
-        match = re.search(r"Filters.Filter.(\d).Name", key)
-        if match:
-            filter_index = match.groups()[0]
-            value_prefix = "Filters.Filter.{0}.Value".format(filter_index)
-            filter_values = [
-                filter_value[0]
-                for filter_key, filter_value in querystring.items()
-                if filter_key.startswith(value_prefix)
-            ]
-            # The AWS query protocol serializes empty lists as an empty string.
-            if filter_values == [""]:
-                filter_values = []
-            response_values[value[0]] = filter_values
-    return response_values
-
-
 def get_object_value(obj, attr):
     """Retrieves an arbitrary attribute value from an object.
 

tests/test_rds/test_utils.py

@@ -4,7 +4,6 @@ from moto.rds.utils import (
     FilterDef,
     apply_filter,
     merge_filters,
-    filters_from_querystring,
     validate_filters,
 )
 
@@ -135,38 +134,3 @@ class TestMergingFilters(object):
         assert len(merged.keys()) == 4
         for key in merged.keys():
             assert merged[key] == ["value1", "value2"]
-
-
-class TestParsingFiltersFromQuerystring(object):
-    def test_parse_empty_list(self):
-        # The AWS query protocol serializes empty lists as an empty string.
-        querystring = {
-            "Filters.Filter.1.Name": ["empty-filter"],
-            "Filters.Filter.1.Value.1": [""],
-        }
-        filters = filters_from_querystring(querystring)
-        assert filters == {"empty-filter": []}
-
-    def test_multiple_values(self):
-        querystring = {
-            "Filters.Filter.1.Name": ["multi-value"],
-            "Filters.Filter.1.Value.1": ["value1"],
-            "Filters.Filter.1.Value.2": ["value2"],
-        }
-        filters = filters_from_querystring(querystring)
-        values = filters["multi-value"]
-        assert len(values) == 2
-        assert "value1" in values
-        assert "value2" in values
-
-    def test_multiple_filters(self):
-        querystring = {
-            "Filters.Filter.1.Name": ["filter-1"],
-            "Filters.Filter.1.Value.1": ["value1"],
-            "Filters.Filter.2.Name": ["filter-2"],
-            "Filters.Filter.2.Value.1": ["value2"],
-        }
-        filters = filters_from_querystring(querystring)
-        assert len(filters.keys()) == 2
-        assert filters["filter-1"] == ["value1"]
-        assert filters["filter-2"] == ["value2"]