mc_511_migrate_endpoints

src/api/endpoints/check/routes.py

@@ -0,0 +1,21 @@
+from fastapi import APIRouter, Depends
+
+from src.api.dependencies import get_async_core
+from src.api.endpoints.check.unique_url.response import CheckUniqueURLResponse
+from src.api.endpoints.check.unique_url.wrapper import check_unique_url_wrapper
+from src.core.core import AsyncCore
+
+check_router = APIRouter(
+    prefix="/check",
+    tags=["check"]
+)
+
+@check_router.get("/unique-url")
+async def check_unique_url(
+    url: str,
+    async_core: AsyncCore = Depends(get_async_core),
+) -> CheckUniqueURLResponse:
+    return await check_unique_url_wrapper(
+        adb_client=async_core.adb_client,
+        url=url
+    )

src/api/endpoints/check/unique_url/response.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+
+
+class CheckUniqueURLResponse(BaseModel):
+    unique_url: bool
+    url_id: int | None

src/api/endpoints/check/unique_url/wrapper.py

@@ -0,0 +1,23 @@
+from src.api.endpoints.check.unique_url.response import CheckUniqueURLResponse
+from src.db.client.async_ import AsyncDatabaseClient
+from src.db.queries.urls_exist.model import URLExistsResult
+from src.db.queries.urls_exist.query import URLsExistInDBQueryBuilder
+from src.util.models.full_url import FullURL
+
+
+async def check_unique_url_wrapper(
+    adb_client: AsyncDatabaseClient,
+    url: str
+) -> CheckUniqueURLResponse:
+    result: URLExistsResult = (await adb_client.run_query_builder(
+        URLsExistInDBQueryBuilder(full_urls=[FullURL(url)])
+    ))[0]
+    if result.exists:
+        return CheckUniqueURLResponse(
+            unique_url=False,
+            url_id=result.url_id
+        )
+    return CheckUniqueURLResponse(
+        unique_url=True,
+        url_id=None
+    )

src/api/endpoints/locations/post/query.py

@@ -0,0 +1,46 @@
+from typing import Any
+
+from sqlalchemy import select, RowMapping
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.api.endpoints.locations.post.request import AddLocationRequestModel
+from src.api.endpoints.locations.post.response import AddLocationResponseModel
+from src.db import Locality, Location
+from src.db.queries.base.builder import QueryBuilderBase
+
+
+class AddLocationQueryBuilder(QueryBuilderBase):
+
+    def __init__(
+        self,
+        request: AddLocationRequestModel
+    ):
+        super().__init__()
+        self.request = request
+
+    async def run(self, session: AsyncSession) -> AddLocationResponseModel:
+        locality = Locality(
+            name=self.request.locality_name,
+            county_id=self.request.county_id
+        )
+        session.add(locality)
+        await session.flush()
+        locality_id: int = locality.id
+
+        query = (
+            select(
+                Location.id
+            )
+            .where(
+                Location.locality_id == locality_id
+            )
+        )
+
+        mapping: RowMapping = await self.sh.mapping(
+            session=session,
+            query=query
+        )
+
+        return AddLocationResponseModel(
+            location_id=mapping[Location.id]
+        )

src/api/endpoints/locations/post/request.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+
+
+class AddLocationRequestModel(BaseModel):
+    locality_name: str
+    county_id: int

src/api/endpoints/locations/post/response.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+
+class AddLocationResponseModel(BaseModel):
+    location_id: int

src/api/endpoints/locations/routes.py

@@ -0,0 +1,22 @@
+from fastapi import APIRouter, Depends
+
+from src.api.dependencies import get_async_core
+from src.api.endpoints.locations.post.query import AddLocationQueryBuilder
+from src.api.endpoints.locations.post.request import AddLocationRequestModel
+from src.api.endpoints.locations.post.response import AddLocationResponseModel
+from src.core.core import AsyncCore
+
+location_url_router = APIRouter(
+    prefix="/locations",
+    tags=["Locations"],
+    responses={404: {"description": "Not found"}}
+)
+
+@location_url_router.post("")
+async def create_location(
+    request: AddLocationRequestModel,
+    async_core: AsyncCore = Depends(get_async_core),
+) -> AddLocationResponseModel:
+    return await async_core.adb_client.run_query_builder(
+        AddLocationQueryBuilder(request)
+    )

src/api/main.py

@@ -11,9 +11,11 @@
 from src.api.endpoints.agencies.routes import agencies_router
 from src.api.endpoints.annotate.routes import annotate_router
 from src.api.endpoints.batch.routes import batch_router
+from src.api.endpoints.check.routes import check_router
 from src.api.endpoints.collector.routes import collector_router
 from src.api.endpoints.contributions.routes import contributions_router
 from src.api.endpoints.data_source.routes import data_sources_router
+from src.api.endpoints.locations.routes import location_url_router
 from src.api.endpoints.meta_url.routes import meta_urls_router
 from src.api.endpoints.metrics.routes import metrics_router
 from src.api.endpoints.root import root_router
@@ -183,7 +185,9 @@ async def redirect_docs():
     contributions_router,
     agencies_router,
     data_sources_router,
-    meta_urls_router
+    meta_urls_router,
+    check_router,
+    location_url_router
 ]
 
 for router in routers:

tests/automated/integration/api/locations/post/test_locality.py

@@ -0,0 +1,38 @@
+import pytest
+
+from src.api.endpoints.locations.post.request import AddLocationRequestModel
+from src.api.endpoints.locations.post.response import AddLocationResponseModel
+from src.db import Locality, Location
+from src.db.client.async_ import AsyncDatabaseClient
+from tests.helpers.api_test_helper import APITestHelper
+from tests.helpers.data_creator.models.creation_info.county import CountyCreationInfo
+
+
+@pytest.mark.asyncio
+async def test_add_locality(
+    allegheny_county: CountyCreationInfo,
+    adb_client_test: AsyncDatabaseClient,
+    api_test_helper: APITestHelper
+):
+    # Add Locality
+    locality_response: dict = api_test_helper.request_validator.post_v3(
+        "/locations",
+        json=AddLocationRequestModel(
+            locality_name="Test Locality",
+            county_id=allegheny_county.county_id
+        ).model_dump(mode='json')
+    )
+    response_model = AddLocationResponseModel(
+        **locality_response
+    )
+
+    # Confirm exists in database
+    localities: list[Locality] = await adb_client_test.get_all(Locality)
+    assert len(localities) == 1
+    assert localities[0].name == "Test Locality"
+    assert localities[0].county_id == allegheny_county.county_id
+
+    locations: list[Location] = await adb_client_test.get_all(Location)
+    assert len(locations) == 3
+    location_ids = {location.id for location in locations}
+    assert response_model.location_id in location_ids

tests/automated/integration/readonly/api/check/test_unique_url.py

@@ -0,0 +1,33 @@
+import pytest
+
+from src.api.endpoints.check.unique_url.response import CheckUniqueURLResponse
+from tests.automated.integration.readonly.helper import ReadOnlyTestHelper
+from tests.helpers.api_test_helper import APITestHelper
+
+
+@pytest.mark.asyncio
+async def test_check_unique_url(
+    readonly_helper: ReadOnlyTestHelper
+):
+
+    ath: APITestHelper = readonly_helper.api_test_helper
+    response_not_unique_url = ath.request_validator.get_v3(
+        url="/check/unique-url",
+        params={
+            "url": "https://read-only-ds.com"
+        }
+    )
+    model_not_unique_url = CheckUniqueURLResponse(**response_not_unique_url)
+    assert not model_not_unique_url.unique_url
+    assert model_not_unique_url.url_id == readonly_helper.maximal_data_source_url_id
+
+
+    response_unique_url = ath.request_validator.get_v3(
+        url="/check/unique-url",
+        params={
+            "url": "https://nonexistent-url.com"
+        }
+    )
+    model_unique_url = CheckUniqueURLResponse(**response_unique_url)
+    assert model_unique_url.unique_url
+    assert model_unique_url.url_id is None

tests/automated/integration/readonly/setup/core.py

@@ -59,9 +59,6 @@ async def setup_readonly_data(
         adb_client=adb_client
     )
 
-
-
-
     # Add Data Source With Linked Agency
     maximal_data_source: int = await add_maximal_data_source(
         agency_1_id=agency_1_id,