Merge pull request #552 from eventcally/issues/551

Add postal code filter to Organizations #551

project/api/organization/schemas.py

@@ -79,6 +79,10 @@ class OrganizationListRequestSchema(
     keyword = fields.Str(
         metadata={"description": "Looks for keyword in name and short name."},
     )
+    postal_code = fields.List(
+        fields.Str(),
+        metadata={"description": "Looks for organizations with this postal code."},
+    )
     sort = fields.Str(
         metadata={"description": "Sort result items."},
         validate=validate.OneOf(

project/services/admin_unit.py

@@ -181,7 +181,7 @@ def get_admin_unit_member(id):
 
 
 def get_admin_unit_query(params: AdminUnitSearchParams):
-    query = AdminUnit.query
+    query = AdminUnit.query.join(AdminUnit.location, isouter=True)
 
     if not params.include_unverified:
         query = query.filter(AdminUnit.is_verified)
@@ -210,6 +210,23 @@ def get_admin_unit_query(params: AdminUnitSearchParams):
 
     query = params.get_trackable_query(query, AdminUnit)
 
+    if params.postal_code:
+        if type(params.postal_code) is list:
+            postalCodes = params.postal_code
+        else:  # pragma: no cover
+            postalCodes = [params.postal_code]
+
+        postalCodeFilters = None
+        for postalCode in postalCodes:
+            postalCodeFilter = Location.postalCode.ilike(postalCode + "%")
+            if postalCodeFilters is not None:
+                postalCodeFilters = or_(postalCodeFilters, postalCodeFilter)
+            else:
+                postalCodeFilters = postalCodeFilter
+
+        if postalCodeFilters is not None:
+            query = query.filter(postalCodeFilters)
+
     if params.keyword:
         like_keyword = "%" + params.keyword + "%"
         order_keyword = params.keyword + "%"

project/services/search_params.py

@@ -20,6 +20,23 @@ class BaseSearchParams(object):
     def load_from_request(self, **kwargs):
         self.sort = kwargs.get("sort", self.sort)
 
+    def load_list_param(self, param: str):
+        item_ids = request.args.getlist(param)
+
+        if len(item_ids) == 1 and "," in item_ids[0]:
+            item_ids = [i.strip() for i in item_ids[0].split(",")]
+
+        if "0" in item_ids:
+            item_ids.remove("0")
+
+        if len(item_ids) > 0:
+            return item_ids
+
+        return None
+
+    def load_bool_param(self, param: str):
+        return request.args[param].lower() in ("true", "t", "yes", "y", "on", "1")
+
 
 class TrackableSearchParams(BaseSearchParams):
     def __init__(self):
@@ -86,12 +103,16 @@ class AdminUnitSearchParams(TrackableSearchParams):
         self.only_verifier = False
         self.reference_request_for_admin_unit_id = None
         self.incoming_verification_requests_postal_code = None
+        self.postal_code = None
 
     def load_from_request(self, **kwargs):
         super().load_from_request(**kwargs)
 
         self.keyword = kwargs.get("keyword", self.keyword)
 
+        if "postal_code" in request.args:
+            self.postal_code = self.load_list_param("postal_code")
+
 
 class OrganizerSearchParams(TrackableSearchParams):
     def __init__(self):
@@ -207,23 +228,6 @@ class EventSearchParams(TrackableSearchParams):
         self.date_from = today
         self.date_to = date_set_end_of_day(today + relativedelta(months=3))
 
-    def load_list_param(self, param: str):
-        item_ids = request.args.getlist(param)
-
-        if len(item_ids) == 1 and "," in item_ids[0]:
-            item_ids = [i.strip() for i in item_ids[0].split(",")]
-
-        if "0" in item_ids:
-            item_ids.remove("0")
-
-        if len(item_ids) > 0:
-            return item_ids
-
-        return None
-
-    def load_bool_param(self, param: str):
-        return request.args[param].lower() in ("true", "t", "yes", "y", "on", "1")
-
     def load_status_list_param(self):
         stati = self.load_list_param("status")
 

tests/api/test_organization.py

@@ -23,6 +23,12 @@ def test_list(client, seeder: Seeder, utils: UtilActions):
     utils.get_json_ok(url)
 
 
+def test_list_postal_code(client, seeder: Seeder, utils: UtilActions):
+    seeder.setup_api_access(user_access=False)
+    url = utils.get_url("api_v1_organization_list", postal_code="38640,38690")
+    utils.get_json_ok(url)
+
+
 def test_list_for_reference_request(client, seeder: Seeder, utils: UtilActions):
     user_id, admin_unit_id = seeder.setup_api_access()
     other_user_id = seeder.create_user("other@test.de")