Add search API functionality

app.py

@@ -1,6 +1,8 @@
 from flask import Flask
-from flask_restful import Api
+from flask_restx import Api
 from flask_cors import CORS
+
+from resources.Search import Search
 from resources.User import User
 from resources.Login import Login
 from resources.RefreshSession import RefreshSession
@@ -41,6 +43,7 @@ def create_app() -> Flask:
         (ResetPassword, "/reset-password"),
         (ResetTokenValidation, "/reset-token-validation"),
         (QuickSearch, "/quick-search/<search>/<location>"),
+        (Search, "/search/<string:coarse_record_type>/<string:location>"),
         (Archives, "/archives"),
         (DataSources, "/data-sources"),
         (DataSourcesMap, "/data-sources-map"),

middleware/search_query.py

@@ -0,0 +1,210 @@
+"""
+This module contains functions and helper functions
+used by the `Search` resource
+"""
+
+import spacy
+import json
+import datetime
+from typing import List, Dict, Any
+from psycopg2.extensions import connection as PgConnection
+
+# TODO: Create search_query_logs table to complement quick_search_query_logs
+
+
+def expand_params(single_param: str, times: int):
+    """
+    Expand the single parameter a given number of times.
+    Used for expanding parameters used repeatedly in SQL parameters
+    :param single_param: The single parameter to be repeated multiple times.
+    :param times: The number of times the single parameter should be repeated.
+    :return: A tuple containing the repeated single_param.
+    """
+    return tuple([single_param] * times)
+
+
+QUICK_SEARCH_COLUMNS = [
+    "airtable_uid",
+    "data_source_name",
+    "description",
+    "record_type",
+    "source_url",
+    "record_format",
+    "coverage_start",
+    "coverage_end",
+    "agency_supplied",
+    "agency_name",
+    "municipality",
+    "state_iso",
+]
+
+QUICK_SEARCH_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
+    WHERE
+        data_sources.record_type = ANY(%s) AND
+        (
+            agencies.county_name LIKE %s OR
+            substr(agencies.county_name,3,length(agencies.county_name)-4)
+                || ' County' LIKE %s
+            OR agencies.state_iso LIKE %s
+            OR agencies.municipality LIKE %s
+            OR agencies.agency_type LIKE %s
+            OR agencies.jurisdiction_type LIKE %s
+            OR agencies.name LIKE %s
+            OR state_names.state_name LIKE %s
+        )
+        AND data_sources.approval_status = 'approved'
+        AND data_sources.url_status not in ('broken', 'none found')
+"""
+
+INSERT_LOG_QUERY = """
+    INSERT INTO search_query_logs
+    (search, location, results, result_count, created_at, datetime_of_request)
+    VALUES (%s, %s, %s, %s, %s, %s)
+"""
+
+
+class SearchQueryEngine:
+    """
+    A search query engine to perform SQL queries
+    for searching records based on a search term and location.
+    """
+
+    def __init__(self, connection: PgConnection):
+        """
+        Setup connection to PostgreSQL database and spacy nlp object
+        :param connection: A PgConnection object
+        """
+        self.conn = connection
+        self.nlp = spacy.load("en_core_web_sm")
+
+    def execute_query(
+        self, coarse_record_types: list[str], location: str
+    ) -> List[Dict[str, Any]]:
+        """
+        Execute a SQL query to search for records
+        based on a search term and location.
+
+        :param search_term: The search term to query for.
+        :param location: The location to search within.
+        :return: A list of dictionaries containing the fetched records.
+        """
+        assert isinstance(
+            coarse_record_types, list
+        ), "coarse_record_types must be a list"
+        with self.conn.cursor() as cursor:
+            cursor.execute(
+                QUICK_SEARCH_SQL, (coarse_record_types,) + expand_params(location, 8)
+            )
+            return cursor.fetchall()
+
+    def print_query_parameters(self, search_terms: list[str], location: str) -> None:
+        """
+        :param search_terms: The search terms used in the query.
+        :param location: The location used in the query.
+        :return: None.
+        """
+        print(f"Query parameters: '%{search_terms}%', '%{location}%'")
+
+    def lemmatize(self, entries: list[str]) -> list[str]:
+        """
+        Lemmatize all entries -- removing inflected forms to better enable searching
+        :param entries: entries to be lemmatized
+        :return: lemmatized results
+        """
+        lemmatized_terms = []
+        for entry in entries:
+            doc = self.nlp(entry.strip())
+            lemmatized_term = " ".join([token.lemma_ for token in doc])
+            lemmatized_terms.append(lemmatized_term)
+        return lemmatized_terms
+
+    def search_query(
+        self, record_types: list[str], location: str
+    ) -> List[Dict[str, Any]]:
+        """
+        Perform a search query based on the given parameters.
+
+        :param record_types: The record types to search for.
+        :param location: The location to search within.
+        :return: A list of dictionaries,
+            where each dictionary represents a search result.
+
+        """
+        self.print_query_parameters(record_types, location)
+        results = self.execute_query(record_types, location)
+        return results
+
+    def quick_search(
+        self,
+        coarse_record_types: list[str] = None,
+        location: str = "",
+    ) -> Dict[str, Any]:
+        """
+        Perform a quick search based on the provided parameters.
+
+        :param coarse_record_types: The types of records to search for.
+        :param location: The location to search for records in.
+        :param test: A flag indicating whether this is a test search.
+        :return: A dictionary containing the search results.
+        """
+        unaltered_results = self.search_query(coarse_record_types, location)
+        spacy_results = self.search_query(
+            record_types=self.lemmatize(coarse_record_types), location=location
+        )
+
+        results = max(spacy_results, unaltered_results, key=len)
+        data_sources = {"count": len(results), "data": results}
+
+        self.log_query_results(coarse_record_types, location, data_sources)
+
+        return data_sources
+
+    def log_query_results(
+        self,
+        coarse_record_types: list[str],
+        location: str,
+        data_sources: Dict[str, Any],
+    ) -> None:
+        datetime_string = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        # If both coarse_record_types and location are blank, all results will have been returned
+        if len(coarse_record_types) > 0 or location != "":
+            query_results = [record["airtable_uid"] for record in data_sources["data"]]
+        else:
+            query_results = ['ALL']
+        with self.conn.cursor() as cursor:
+            cursor.execute(
+                INSERT_LOG_QUERY,
+                (
+                    coarse_record_types,
+                    location,
+                    json.dumps(query_results),
+                    data_sources["count"],
+                    datetime_string,
+                ),
+            )
+            self.conn.commit()
+

requirements.txt

@@ -20,6 +20,7 @@ exceptiongroup==1.1.3
 Flask==2.3.2
 Flask-Cors==4.0.0
 Flask-RESTful==0.3.10
+flask-restx==1.3.0
 gotrue==1.0.3
 gunicorn==21.2.0
 h11==0.14.0
@@ -46,7 +47,7 @@ pycparser==2.21
 pydantic==2.2.1
 pydantic_core==2.6.1
 PyJWT==2.8.0
-pytest==6.2.5
+pytest~=8.1.2
 python-dateutil==2.8.2
 python-dotenv==1.0.0
 pytz==2023.3
@@ -75,4 +76,5 @@ wasabi==1.1.2
 websockets==10.4
 Werkzeug==3.0.1
 zipp==3.16.2
-pytest-mock~=3.12.0
+pytest-mock~=3.12.0
+lorem~=0.1.1

resources/PsycopgResource.py

@@ -1,12 +1,13 @@
-from flask_restful import Resource
+from flask_restx import Resource
 
 
 class PsycopgResource(Resource):
-    def __init__(self, **kwargs):
+    def __init__(self, *args, **kwargs):
         """
         Initializes the resource with a database connection.
         - kwargs (dict): Keyword arguments containing 'psycopg2_connection' for database connection.
         """
+        super().__init__(*args, **kwargs)
         self.psycopg2_connection = kwargs["psycopg2_connection"]
 
     def get(self):

resources/Search.py

@@ -0,0 +1,84 @@
+from middleware.search_query import SearchQueryEngine
+from middleware.security import api_required
+import requests
+import json
+import os
+from middleware.initialize_psycopg2_connection import initialize_psycopg2_connection
+from flask import request
+from typing import Dict, Any
+
+from resources.PsycopgResource import PsycopgResource
+
+
+class Search(PsycopgResource):
+    """
+    Provides a resource for performing quick searches in the database for data sources
+    based on user-provided search terms and location.
+    """
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.engine = SearchQueryEngine(connection=self.psycopg2_connection)
+
+    # 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
+    def get(self, coarse_record_types: str, location: str) -> Dict[str, Any]:
+        """
+        Performs a quick search using the provided search terms and location. It attempts to find relevant
+        data sources in the database. If no results are found initially, it re-initializes the database
+        connection and tries again.
+
+        Parameters:
+        - search (str): The search term provided by the user.
+        - location (str): The location provided by the user.
+
+        Returns:
+        - A dictionary containing a message about the search results and the data found, if any.
+        """
+        try:
+            data = request.get_json()
+            test = data.get("test_flag")
+        except:
+            test = False
+
+        if isinstance(coarse_record_types, str):
+            course_record_types = [coarse_record_types]
+        try:
+            data_sources = self.engine.quick_search(course_record_types, location, test)
+
+            if data_sources["count"] == 0:
+                self.psycopg2_connection = initialize_psycopg2_connection()
+                data_sources = self.engine.quick_search(course_record_types, location)
+
+            if data_sources["count"] == 0:
+                return {
+                    "count": 0,
+                    "message": "No results found. Please considering requesting a new data source.",
+                }, 404
+
+            return {
+                "message": "Results for search successfully retrieved",
+                "data": data_sources,
+            }
+
+        except Exception as e:
+            self.psycopg2_connection.rollback()
+            print(str(e))
+            webhook_url = os.getenv("WEBHOOK_URL")
+            user_message = "There was an error during the search operation"
+            message = {
+                "content": user_message
+                + ": "
+                + str(e)
+                + "\n"
+                + f"Record Types: {course_record_types}\n"
+                + f"Location: {location}"
+            }
+            requests.post(
+                webhook_url,
+                data=json.dumps(message),
+                headers={"Content-Type": "application/json"},
+            )
+
+            return {"count": 0, "message": user_message}, 500