Add logic for synchronizing followed locations from Data Sources App

pyproject.toml

@@ -23,7 +23,7 @@ dependencies = [
     "lxml~=5.1.0",
     "marshmallow~=3.23.2",
     "openai~=1.60.1",
-    "pdap-access-manager==0.3.6",
+    "pdap-access-manager==0.4.2",
     "pillow>=11.3.0",
     "pip>=25.2",
     "playwright~=1.49.1",

src/external/pdap/client.py

@@ -7,6 +7,8 @@
 from src.external.pdap.dtos.match_agency.response import MatchAgencyResponse
 from src.external.pdap.dtos.unique_url_duplicate import UniqueURLDuplicateInfo
 from src.external.pdap.enums import MatchAgencyResponseStatus
+from src.external.pdap.impl.follows.core import get_user_followed_locations
+from src.external.pdap.impl.follows.response import GetFollowsResponse
 from src.external.pdap.impl.meta_urls.core import submit_meta_urls
 from src.external.pdap.impl.meta_urls.request import SubmitMetaURLsRequest
 
@@ -156,4 +158,9 @@
         return await submit_meta_urls(
             self.access_manager,
             requests=requests
-        )
+        )
+
+    async def get_user_followed_locations(self) -> GetFollowsResponse:
+        return await get_user_followed_locations(
+            access_manager=self.access_manager
+        )

src/external/pdap/impl/follows/core.py

@@ -0,0 +1,20 @@
+from pdap_access_manager import AccessManager, DataSourcesNamespaces, RequestInfo, RequestType, ResponseInfo
+
+from src.external.pdap.impl.follows.response import GetFollowsResponse, LinkUserFollow
+
+
+async def get_user_followed_locations(
+    access_manager: AccessManager,
+) -> GetFollowsResponse:
+
+    url: str = f"{access_manager.data_sources_url}/v3/v2/source-collector/follows"
+    headers: dict[str, str] = await access_manager.jwt_header()
+    request_info = RequestInfo(
+        type_=RequestType.GET,
+        url=url,
+        headers=headers
+    )
+    response_info: ResponseInfo = await access_manager.make_request(request_info)
+    return GetFollowsResponse(
+        **response_info.data
+    )

src/external/pdap/impl/follows/response.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel
+
+
+class LinkUserFollow(BaseModel):
+    user_id: int
+    location_id: int
+
+
+class GetFollowsResponse(BaseModel):
+    follows: list[LinkUserFollow]
+

src/security/manager.py

@@ -68,8 +68,3 @@ def get_access_info(
         token: Annotated[str, Depends(oauth2_scheme)]
 ) -> AccessInfo:
     return SecurityManager().check_access(token, Permissions.SOURCE_COLLECTOR)
-
-def require_permission(permission: Permissions):
-    def dependency(token: Annotated[str, Depends(oauth2_scheme)]) -> AccessInfo:
-        return SecurityManager().check_access(token, permission=permission)
-    return dependency

tests/manual/external/pdap/test_get_follows.py

@@ -0,0 +1,7 @@
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_get_user_followed_locations(pdap_client_dev):
+    response = await pdap_client_dev.get_user_followed_locations()
+    print(response)

tests/manual/external/pdap/test_match_agency.py

@@ -4,3 +4,4 @@
 @pytest.mark.asyncio
 async def test_match_agency(pdap_client):
     response = await pdap_client.match_agency(name="police")
+    print(response)