Add search_fraud_checks endpoint to FlashpointConnector

dev_env/api/test_flashpoint.py

@@ -1,32 +1,16 @@
-from pyapiary.api_connectors.flashpoint import FlashpointConnector
+import asyncio
 import httpx
+from pyapiary.api_connectors.flashpoint import FlashpointConnector, AsyncFlashpointConnector
 
-fp = FlashpointConnector(
+fp = AsyncFlashpointConnector(
     load_env_vars=True,
     trust_env=True,
     verify=False,
     follow_redirects=True
 )
 
-# res = fp.search_communities('telegram')
-# print(res)
+async def main():
+    # Insert test here
+    pass
 
-
-try:
-    resp = fp.get_media_image(
-        "gs://kraken-datalake-media/artifacts/67/6739eed871575b5ad1864ea66d42ebf2430eda5f34d40715c9e40efe90232aa6",
-        headers={},
-        request_kwargs={"follow_redirects": True},
-    )
-    print("Final URL:", resp.url)
-    print("Final status:", resp.status_code)
-except httpx.HTTPStatusError as e:
-    resp = e.response
-    print("Final error:", resp.status_code, resp.url)
-
-    # Show the chain of redirects
-    for hop in resp.history:
-        print("Redirect:", hop.status_code, hop.url, "->", hop.headers.get("location"))
-
-    # Show body of final error if any
-    print("Response body:", resp.text[:500])
+asyncio.run(main())

dev_env/api/test_ipqs.py

@@ -17,7 +17,7 @@ async def main():
     )
 
     res = await ipqs.malicious_url(
-        query="cracked.to"
+        query="bad.url"
     )
 
     print(res)

dev_env/api/test_urlscan.py

@@ -18,7 +18,7 @@
 
 
 # ---------- sync ----------
-res = urlscan.search("bclubs.cc")
+res = urlscan.search("google.com")
 print(res.status_code)
 
 res = urlscan.scan("google.com")

src/pyapiary/api_connectors/flashpoint.py

@@ -44,6 +44,16 @@ def search_fraud(self, query: str, **kwargs) -> httpx.Response:
         """
         return self.post("/sources/v2/fraud", json={"query": query, **kwargs})
 
+    @log_method_call
+    def search_fraud_checks(self, query: str, **kwargs) -> httpx.Response:
+        """Checks search provides the ability to search and read data from our Checks dataset.
+
+        Args:
+            query (str): The search string used in the API query.
+            **kwargs: Additional query logic per the Flashpoint API documentation.
+        """
+        return self.post("/sources/v2/fraud/checks", json={"query": query, **kwargs})
+
     @log_method_call
     def search_marketplaces(self, query: str, **kwargs) -> httpx.Response:
         """
@@ -140,6 +150,16 @@ async def search_fraud(self, query: str, **kwargs) -> httpx.Response:
         """
         return await self.post("/sources/v2/fraud", json={"query": query, **kwargs})
 
+    @log_method_call
+    async def search_fraud_checks(self, query: str, **kwargs) -> httpx.Response:
+        """Checks search provides the ability to search and read data from our Checks dataset.
+
+        Args:
+            query (str): The search string used in the API query.
+            **kwargs: Additional query logic per the Flashpoint API documentation.
+        """
+        return await self.post("/sources/v2/fraud/checks", json={"query": query, **kwargs})
+
     @log_method_call
     async def search_marketplaces(self, query: str, **kwargs) -> httpx.Response:
         """

src/pyapiary/tests/test_flashpoint/test_unit_async_flashpoint.py

@@ -26,6 +26,57 @@ async def test_async_init_missing_key():
             AsyncFlashpointConnector()
 
 
+@patch("pyapiary.api_connectors.flashpoint.AsyncFlashpointConnector.post", new_callable=AsyncMock)
+@pytest.mark.asyncio
+async def test_async_search_fraud_checks(mock_post):
+    import json
+
+    request = httpx.Request("POST", "https://api.flashpoint.io/mock")
+    payload = {"success": True, "data": []}
+    mock_response = httpx.Response(
+        200,
+        request=request,
+        content=json.dumps(payload).encode("utf-8"),
+        headers={"Content-Type": "application/json"},
+    )
+    mock_post.return_value = mock_response
+
+    connector = AsyncFlashpointConnector(api_key="mock_token")
+    result = await connector.search_fraud_checks("stolen checks")
+
+    assert isinstance(result, httpx.Response)
+    assert result.json() == payload
+    mock_post.assert_awaited_once_with(
+        "/sources/v2/fraud/checks", json={"query": "stolen checks"}
+    )
+
+
+@patch("pyapiary.api_connectors.flashpoint.AsyncFlashpointConnector.post", new_callable=AsyncMock)
+@pytest.mark.asyncio
+async def test_async_search_fraud_checks_with_kwargs(mock_post):
+    import json
+
+    request = httpx.Request("POST", "https://api.flashpoint.io/mock")
+    payload = {"success": True, "data": [{"id": "abc123"}]}
+    mock_response = httpx.Response(
+        200,
+        request=request,
+        content=json.dumps(payload).encode("utf-8"),
+        headers={"Content-Type": "application/json"},
+    )
+    mock_post.return_value = mock_response
+
+    connector = AsyncFlashpointConnector(api_key="mock_token")
+    result = await connector.search_fraud_checks("routing number", size=10, from_=0)
+
+    assert isinstance(result, httpx.Response)
+    assert result.json() == payload
+    mock_post.assert_awaited_once_with(
+        "/sources/v2/fraud/checks",
+        json={"query": "routing number", "size": 10, "from_": 0},
+    )
+
+
 @patch("pyapiary.api_connectors.flashpoint.AsyncFlashpointConnector.post", new_callable=AsyncMock)
 @pytest.mark.asyncio
 async def test_async_search_fraud(mock_post):

src/pyapiary/tests/test_flashpoint/test_unit_flashpoint.py

@@ -22,6 +22,55 @@ def test_init_missing_key():
             FlashpointConnector()
 
 
+@patch("pyapiary.api_connectors.flashpoint.FlashpointConnector.post")
+def test_search_fraud_checks(mock_post):
+    import json
+
+    request = httpx.Request("POST", "https://api.flashpoint.io/mock")
+    payload = {"success": True, "data": []}
+    mock_response = httpx.Response(
+        200,
+        request=request,
+        content=json.dumps(payload).encode("utf-8"),
+        headers={"Content-Type": "application/json"},
+    )
+    mock_post.return_value = mock_response
+
+    connector = FlashpointConnector(api_key="mock_token")
+    result = connector.search_fraud_checks("stolen checks")
+
+    assert isinstance(result, httpx.Response)
+    assert result.json() == payload
+    mock_post.assert_called_once_with(
+        "/sources/v2/fraud/checks", json={"query": "stolen checks"}
+    )
+
+
+@patch("pyapiary.api_connectors.flashpoint.FlashpointConnector.post")
+def test_search_fraud_checks_with_kwargs(mock_post):
+    import json
+
+    request = httpx.Request("POST", "https://api.flashpoint.io/mock")
+    payload = {"success": True, "data": [{"id": "abc123"}]}
+    mock_response = httpx.Response(
+        200,
+        request=request,
+        content=json.dumps(payload).encode("utf-8"),
+        headers={"Content-Type": "application/json"},
+    )
+    mock_post.return_value = mock_response
+
+    connector = FlashpointConnector(api_key="mock_token")
+    result = connector.search_fraud_checks("routing number", size=10, from_=0)
+
+    assert isinstance(result, httpx.Response)
+    assert result.json() == payload
+    mock_post.assert_called_once_with(
+        "/sources/v2/fraud/checks",
+        json={"query": "routing number", "size": 10, "from_": 0},
+    )
+
+
 @patch("pyapiary.api_connectors.flashpoint.FlashpointConnector.post")
 def test_search_fraud(mock_post):
     import json