Create /search/agency endpoint with test

src/api/endpoints/search/agency/models/response.py

@@ -0,0 +1,7 @@
+from pydantic import BaseModel
+
+
+class AgencySearchResponse(BaseModel):
+    agency_id: int
+    agency_name: str
+    location_display_name: str

src/api/endpoints/search/agency/query.py

@@ -0,0 +1,66 @@
+from typing import Sequence
+
+from sqlalchemy import select, func, RowMapping
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.api.endpoints.search.agency.models.response import AgencySearchResponse
+from src.db.helpers.session import session_helper as sh
+from src.db.models.impl.agency.sqlalchemy import Agency
+from src.db.models.impl.link.agency_location.sqlalchemy import LinkAgencyLocation
+from src.db.models.views.location_expanded import LocationExpandedView
+from src.db.queries.base.builder import QueryBuilderBase
+
+
+class SearchAgencyQueryBuilder(QueryBuilderBase):
+
+    def __init__(
+        self,
+        location_id: int | None,
+        query: str | None
+    ):
+        super().__init__()
+        self.location_id = location_id
+        self.query = query
+
+    async def run(self, session: AsyncSession) -> list[AgencySearchResponse]:
+
+        query = (
+            select(
+                Agency.agency_id,
+                Agency.name.label("agency_name"),
+                LocationExpandedView.display_name.label("location_display_name")
+            )
+            .join(
+                LinkAgencyLocation,
+                LinkAgencyLocation.agency_id == Agency.agency_id
+            )
+            .join(
+                LocationExpandedView,
+                LocationExpandedView.id == LinkAgencyLocation.location_id
+            )
+        )
+
+        if self.location_id is not None:
+            query = query.where(
+                LocationExpandedView.id == self.location_id
+            )
+        if self.query is not None:
+            query = query.order_by(
+                func.similarity(
+                    Agency.name,
+                    self.query
+                ).desc()
+            )
+
+        mappings: Sequence[RowMapping] = await sh.mappings(session, query)
+
+        return [
+            AgencySearchResponse(
+                **mapping
+            )
+            for mapping in mappings
+        ]
+
+
+
+

src/api/endpoints/search/routes.py

@@ -1,6 +1,10 @@
-from fastapi import APIRouter, Query, Depends
+
+from fastapi import APIRouter, Query, Depends, HTTPException
+from starlette import status
 
 from src.api.dependencies import get_async_core
+from src.api.endpoints.search.agency.models.response import AgencySearchResponse
+from src.api.endpoints.search.agency.query import SearchAgencyQueryBuilder
 from src.api.endpoints.search.dtos.response import SearchURLResponse
 from src.core.core import AsyncCore
 from src.security.manager import get_access_info
@@ -18,4 +22,31 @@
     """
     Search for a URL in the database
     """
-    return await async_core.search_for_url(url)
+    return await async_core.search_for_url(url)
+
+
+@search_router.get("/agency")
+async def search_agency(
+    location_id: int | None = Query(
+        description="The location id to search for",
+        default=None
+    ),
+    query: str | None = Query(
+        description="The query to search for",
+        default=None
+    ),
+    access_info: AccessInfo = Depends(get_access_info),
+    async_core: AsyncCore = Depends(get_async_core),
+) -> list[AgencySearchResponse]:
+    if query is None and location_id is None:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="At least one of query or location_id must be provided"
+        )
+
+    return await async_core.adb_client.run_query_builder(
+        SearchAgencyQueryBuilder(
+            location_id=location_id,
+            query=query
+        )
+    )

tests/automated/integration/api/search/agency/test_search.py

@@ -0,0 +1,53 @@
+import pytest
+
+from tests.helpers.api_test_helper import APITestHelper
+from tests.helpers.data_creator.core import DBDataCreator
+from tests.helpers.data_creator.models.creation_info.county import CountyCreationInfo
+from tests.helpers.data_creator.models.creation_info.locality import LocalityCreationInfo
+
+
+@pytest.mark.asyncio
+async def test_search_agency(
+    api_test_helper: APITestHelper,
+    db_data_creator: DBDataCreator,
+    pittsburgh_locality: LocalityCreationInfo,
+    allegheny_county: CountyCreationInfo
+):
+
+    agency_a_id: int = await db_data_creator.agency("A Agency")
+    agency_b_id: int = await db_data_creator.agency("AB Agency")
+    agency_c_id: int = await db_data_creator.agency("ABC Agency")
+
+    await db_data_creator.link_agencies_to_location(
+        agency_ids=[agency_a_id, agency_c_id],
+        location_id=pittsburgh_locality.location_id
+    )
+    await db_data_creator.link_agencies_to_location(
+        agency_ids=[agency_b_id],
+        location_id=allegheny_county.location_id
+    )
+
+    responses: list[dict] = api_test_helper.request_validator.get_v2(
+        url="/search/agency",
+        params={
+            "query": "A Agency",
+        }
+    )
+    assert len(responses) == 3
+    assert responses[0]["agency_id"] == agency_a_id
+    assert responses[1]["agency_id"] == agency_b_id
+    assert responses[2]["agency_id"] == agency_c_id
+
+    responses = api_test_helper.request_validator.get_v2(
+        url="/search/agency",
+        params={
+            "query": "A Agency",
+            "location_id": pittsburgh_locality.location_id
+        }
+    )
+
+    assert len(responses) == 2
+    assert responses[0]["agency_id"] == agency_a_id
+    assert responses[1]["agency_id"] == agency_c_id
+
+

tests/helpers/data_creator/commands/impl/agency.py

@@ -6,10 +6,21 @@
 from src.core.enums import SuggestionType
 from src.core.tasks.url.operators.agency_identification.dtos.suggestion import URLAgencySuggestionInfo
 from tests.helpers.data_creator.commands.base import DBDataCreatorCommandBase
+from tests.helpers.simple_test_data_functions import generate_test_name
+
 
 @final
 class AgencyCommand(DBDataCreatorCommandBase):
 
+    def __init__(
+        self,
+        name: str | None = None
+    ):
+        super().__init__()
+        if name is None:
+            name = generate_test_name()
+        self.name = name
+
     @override
     async def run(self) -> int:
         agency_id = randint(1, 99999999)
@@ -19,7 +30,7 @@
                     url_id=-1,
                     suggestion_type=SuggestionType.UNKNOWN,
                     pdap_agency_id=agency_id,
-                    agency_name=f"Test Agency {agency_id}",
+                    agency_name=self.name,
                     state=f"Test State {agency_id}",
                     county=f"Test County {agency_id}",
                     locality=f"Test Locality {agency_id}"

tests/helpers/data_creator/core.py

@@ -156,8 +156,8 @@
             urls=[iui.url for iui in iuis.url_mappings]
         )
 
-    async def agency(self) -> int:
-        return await self.run_command(AgencyCommand())
+    async def agency(self, name: str | None = None) -> int:
+        return await self.run_command(AgencyCommand(name))
 
     async def auto_relevant_suggestions(self, url_id: int, relevant: bool = True):
         await self.run_command(

tests/helpers/simple_test_data_functions.py

@@ -4,6 +4,8 @@
 """
 import uuid
 
+from tests.helpers.counter import next_int
+
 
 def generate_test_urls(count: int) -> list[str]:
     results = []
@@ -17,7 +19,9 @@
 def generate_test_url(i: int) -> str:
     return f"https://test.com/{i}"
 
-def generate_test_name(i: int) -> str:
+def generate_test_name(i: int | None = None) -> str:
+    if i is None:
+        return f"Test Name {next_int()}"
     return f"Test Name {i}"
 
 def generate_test_description(i: int) -> str: