Mc issue 352 search endpoint

app.py

@@ -3,6 +3,7 @@
 from flask import Flask
 from flask_cors import CORS
 
+from resources.Search import namespace_search
 from resources.TypeaheadSuggestions import (
     TypeaheadSuggestions,
     namespace_typeahead_suggestions,
@@ -38,6 +39,7 @@
     namespace_reset_password,
     namespace_quick_search,
     namespace_typeahead_suggestions,
+    namespace_search
 ]
 
 MY_PREFIX = "/api"

database_client/database_client.py

@@ -14,6 +14,7 @@
     TokenNotFoundError,
     AccessTokenNotFoundError,
 )
+from utilities.enums import RecordCategories
 
 DATA_SOURCES_MAP_COLUMN = [
     "data_source_id",
@@ -657,3 +658,42 @@
             )
             for row in results
         ]
+
+    def search_with_location_and_record_type(
+        self,
+        state: str,
+        record_type: Optional[RecordCategories] = None,
+        county: Optional[str] = None,
+        locality: Optional[str] = None,
+    ) -> List[QuickSearchResult]:
+        """
+        Searches for data sources in the database.
+
+        :param state: The state to search for data sources in.
+        :param record_type: The type of data sources to search for. If None, all data sources will be searched for.
+        :param county: The county to search for data sources in. If None, all data sources will be searched for.
+        :param locality: The locality to search for data sources in. If None, all data sources will be searched for.
+        :return: A list of QuickSearchResult objects.
+        """
+        query = DynamicQueryConstructor.create_search_query(
+            state=state, record_type=record_type, county=county, locality=locality
+        )
+        self.cursor.execute(query)
+        results = self.cursor.fetchall()
+        return [
+            self.QuickSearchResult(
+                id=row[0],
+                data_source_name=row[1],
+                description=row[2],
+                record_type=row[3],
+                url=row[4],
+                format=row[5],
+                coverage_start=row[6],
+                coverage_end=row[7],
+                agency_supplied=row[8],
+                agency_name=row[9],
+                municipality=row[10],
+                state=row[11],
+            )
+            for row in results
+        ]

database_client/dynamic_query_constructor.py

@@ -11,6 +11,7 @@
     RESTRICTED_DATA_SOURCE_COLUMNS,
     RESTRICTED_COLUMNS,
 )
+from utilities.enums import RecordCategories
 
 TableColumn = namedtuple("TableColumn", ["table", "column"])
 TableColumnAlias = namedtuple("TableColumnAlias", ["table", "column", "alias"])
@@ -266,3 +267,76 @@
             search_term_anywhere=sql.Literal(f"%{search_term}%"),
         )
         return query
+
+    @staticmethod
+    def create_search_query(
+        state: str,
+        record_type: Optional[RecordCategories] = None,
+        county: Optional[str] = None,
+        locality: Optional[str] = None
+    ) ->sql.Composed:
+
+        base_query = sql.SQL("""
+            SELECT
+                data_sources.airtable_uid,
+                data_sources.name AS data_source_name,
+                data_sources.description,
+                data_sources.record_type,
+                data_sources.source_url,
+                data_sources.record_format,
+                data_sources.coverage_start,
+                data_sources.coverage_end,
+                data_sources.agency_supplied,
+                agencies.name AS agency_name,
+                agencies.municipality,
+                agencies.state_iso
+            FROM
+                agency_source_link
+            INNER JOIN
+                data_sources ON agency_source_link.airtable_uid = data_sources.airtable_uid
+            INNER JOIN
+                agencies ON agency_source_link.agency_described_linked_uid = agencies.airtable_uid
+            INNER JOIN
+                state_names ON agencies.state_iso = state_names.state_iso
+            INNER JOIN
+                counties ON agencies.county_fips = counties.fips
+        """)
+
+        join_conditions = []
+        where_conditions = [
+            sql.SQL("state_names.state_name = {state_name}").format(state_name=sql.Literal(state)),
+            sql.SQL("data_sources.approval_status = 'approved'"),
+            sql.SQL("data_sources.url_status NOT IN ('broken', 'none found')")
+        ]
+
+        if record_type is not None:
+            join_conditions.append(sql.SQL("""
+                INNER JOIN
+                    record_types ON data_sources.record_type_id = record_types.id
+                INNER JOIN
+                    record_categories ON record_types.category_id = record_categories.id
+            """))
+
+            where_conditions.append(sql.SQL(
+                "record_categories.name = {record_type}"
+            ).format(record_type=sql.Literal(record_type.value)))
+
+        if county is not None:
+            where_conditions.append(sql.SQL(
+                "counties.name = {county_name}"
+            ).format(county_name=sql.Literal(county)))
+
+        if locality is not None:
+            where_conditions.append(sql.SQL(
+                "agencies.municipality = {locality}"
+            ).format(locality=sql.Literal(locality)))
+
+        query = sql.Composed([
+            base_query,
+            sql.SQL(' ').join(join_conditions),
+            sql.SQL(" WHERE "),
+            sql.SQL(' AND ').join(where_conditions)
+        ])
+
+        return query
+

database_client/result_formatter.py

@@ -1,3 +1,4 @@
+from collections import namedtuple
 from typing import Any
 
 from database_client.constants import (
@@ -68,3 +69,6 @@
         return ResultFormatter.convert_data_source_matches(
             data_source_and_agency_columns, [results]
         )[0]
+
+def dictify_namedtuple(result: list[namedtuple]) -> list[dict[str, Any]]:
+    return [result._asdict() for result in result]

middleware/search_logic.py

@@ -0,0 +1,28 @@
+from http import HTTPStatus
+from typing import Optional
+
+from flask import Response, make_response
+
+from database_client.database_client import DatabaseClient
+from database_client.result_formatter import ResultFormatter, dictify_namedtuple
+from utilities.enums import RecordCategories
+
+
+def search_wrapper(
+    db_client: DatabaseClient,
+    state: str,
+    record_category: Optional[RecordCategories] = None,
+    county: Optional[str] = None,
+    locality: Optional[str] = None,
+) -> Response:
+    search_results = db_client.search_with_location_and_record_type(
+        state=state, record_type=record_category, county=county, locality=locality
+    )
+
+    dict_results = dictify_namedtuple(search_results)
+    # TODO: This is shared by other search routes such as quick-search. Consolidate i
+    body = {
+        "count": len(dict_results),
+        "data": dict_results
+    }
+    return make_response(body, HTTPStatus.OK)

resources/QuickSearch.py

@@ -4,37 +4,19 @@
 from middleware.security import api_required
 from middleware.quick_search_query import quick_search_query_wrapper
 from resources.DataSources import namespace_data_source
-from resources.resource_helpers import add_api_key_header_arg
+from resources.resource_helpers import add_api_key_header_arg, create_search_model
 
 from utilities.namespace import create_namespace
 from resources.PsycopgResource import PsycopgResource
 
 namespace_quick_search = create_namespace()
 
-data_item_model = namespace_quick_search.model('DataItem', {
-    'airtable_uid': fields.String(required=True, description='Airtable UID of the record'),
-    'agency_name': fields.String(description='Name of the agency'),
-    'municipality': fields.String(description='Name of the municipality'),
-    'state_iso': fields.String(description='ISO code of the state'),
-    'data_source_name': fields.String(description='Name of the data source'),
-    'description': fields.String(description='Description of the record'),
-    'record_type': fields.String(description='Type of the record'),
-    'source_url': fields.String(description='URL of the data source'),
-    'record_format': fields.String(description='Format of the record'),
-    'coverage_start': fields.String(description='Coverage start date'),
-    'coverage_end': fields.String(description='Coverage end date'),
-    'agency_supplied': fields.String(description='If the record is supplied by the agency')
-})
-
-# Define the main model
-main_model = namespace_quick_search.model('MainModel', {
-    'count': fields.Integer(required=True, description='Count of data items', attribute='count'),
-    'data': fields.List(fields.Nested(data_item_model, required=True, description='List of data items'), attribute='data')
-})
+search_result_outer_model = create_search_model(namespace_quick_search)
 
 authorization_parser = namespace_quick_search.parser()
 add_api_key_header_arg(authorization_parser)
 
+
 @namespace_quick_search.route("/quick-search/<search>/<location>")
 class QuickSearch(PsycopgResource):
     """
@@ -45,7 +27,7 @@ class QuickSearch(PsycopgResource):
     # api_required decorator requires the request"s header to include an "Authorization" key with the value formatted as "Bearer [api_key]"
     # A user can get an API key by signing up and logging in (see User.py)
     @api_required
-    @namespace_quick_search.response(200, "Success", main_model)
+    @namespace_quick_search.response(200, "Success", search_result_outer_model)
     @namespace_data_source.response(500, "Internal server error")
     @namespace_data_source.response(400, "Bad request; missing or bad API key")
     @namespace_data_source.response(403, "Forbidden; invalid API key")
@@ -55,13 +37,13 @@ class QuickSearch(PsycopgResource):
     @namespace_quick_search.expect(authorization_parser)
     @namespace_quick_search.param(
         name="search",
-        description="The search term provided by the user. Checks partial matches on any of the following properties on the data_source table: \"name\", \"description\", \"record_type\", and \"tags\". The search term is case insensitive and will match singular and pluralized versions of the term.",
-        _in="path"
+        description='The search term provided by the user. Checks partial matches on any of the following properties on the data_source table: "name", "description", "record_type", and "tags". The search term is case insensitive and will match singular and pluralized versions of the term.',
+        _in="path",
     )
     @namespace_quick_search.param(
         name="location",
-        description="The location provided by the user. Checks partial matches on any of the following properties on the agencies table: \"county_name\", \"state_iso\", \"municipality\", \"agency_type\", \"jurisdiction_type\", \"name\"",
-        _in="path"
+        description='The location provided by the user. Checks partial matches on any of the following properties on the agencies table: "county_name", "state_iso", "municipality", "agency_type", "jurisdiction_type", "name"',
+        _in="path",
     )
     def get(self, search: str, location: str) -> Response:
         """

resources/Search.py

@@ -0,0 +1,100 @@
+from flask import Response, request
+
+from middleware.search_logic import search_wrapper
+from middleware.security import api_required
+from resources.PsycopgResource import PsycopgResource
+from resources.resource_helpers import add_api_key_header_arg, create_search_model
+from utilities.enums import RecordCategories
+from utilities.namespace import create_namespace, AppNamespaces
+
+namespace_search = create_namespace(namespace_attributes=AppNamespaces.SEARCH)
+
+request_parser = namespace_search.parser()
+add_api_key_header_arg(request_parser)
+request_parser.add_argument(
+    "state",
+    type=str,
+    location="args",
+    required=True,
+    help="The state of the search. Must be an exact match.",
+)
+
+request_parser.add_argument(
+    "county",
+    type=str,
+    location="args",
+    required=False,
+    help="The county of the search. If empty, all counties for the given state will be searched. Must be an exact "
+         "match.",
+)
+
+request_parser.add_argument(
+    "locality",
+    type=str,
+    location="args",
+    required=False,
+    help="The locality of the search. If empty, all localities for the given county will be searched. Must be an "
+         "exact match.",
+)
+
+request_parser.add_argument(
+    "record_category",
+    type=str,
+    location="args",
+    required=False,
+    help="The record category of the search. If empty, all categories will be searched. Must be an exact match.",
+)
+
+search_model = create_search_model(namespace_search)
+
+
+@namespace_search.route("/search-location-and-record-type")
+class Search(PsycopgResource):
+    """
+    Provides a resource for performing searches in the database for data sources
+    based on user-provided search terms and location.
+    """
+
+    @api_required
+    @namespace_search.expect(request_parser)
+    @namespace_search.response(200, "Success", search_model)
+    @namespace_search.response(500, "Internal server error")
+    @namespace_search.response(400, "Bad request; missing or bad API key")
+    @namespace_search.response(403, "Forbidden; invalid API key")
+    def get(self) -> Response:
+        """
+        Performs a search using the provided search terms and location.
+
+        Performs a search using the provided record type and location parameters.
+        It attempts to find relevant data sources in the database.
+
+        Record Types:
+        - "Police & Public Interactions"
+        - "Info about Officers"
+        - "Info about Agencies"
+        - "Agency-published Resources"
+        - "Jails & Courts"
+
+        Source of truth for record types can be found at https://app.gitbook.com/o/-MXypK5ySzExtEzQU6se/s/-MXyolqTg_voOhFyAcr-/activities/data-dictionaries/record-types-taxonomy
+
+        Returns:
+        - A dictionary containing a message about the search results and the data found, if any.
+        """
+        state = request.args.get("state")
+        county = request.args.get("county")
+        locality = request.args.get("locality")
+        record_category_raw = request.args.get("record_category")
+        if record_category_raw is not None:
+            record_category = RecordCategories(record_category_raw)
+        else:
+            record_category = None
+
+        with self.setup_database_client() as db_client:
+            response = search_wrapper(
+                db_client=db_client,
+                record_category=record_category,
+                state=state,
+                county=county,
+                locality=locality,
+            )
+        return response