Refactor: Move code examples to example/, handle API format compat

Author: Chocapikk

README.md

@@ -13,48 +13,6 @@ pip install leakix
 
 To run tests, use `poetry run pytest`.
 
-## Quick Start
-
-```python
-from leakix import Client
-
-client = Client(api_key="your-api-key")
-
-# Simple search - same syntax as the website
-results = client.search("+plugin:GitConfigHttpPlugin", scope="leak")
-for event in results.json():
-    print(event.ip, event.host)
-
-# Search services
-results = client.search("+country:FR +port:22", scope="service")
-```
-
-## Async Client
-
-For async applications, use `AsyncClient`:
-
-```python
-import asyncio
-from leakix import AsyncClient
-
-async def main():
-    async with AsyncClient(api_key="your-api-key") as client:
-        # Simple search
-        results = await client.search("+plugin:GitConfigHttpPlugin", scope="leak")
-        for event in results:
-            print(event.ip, event.host)
-
-        # Host lookup
-        host = await client.get_host("8.8.8.8")
-        print(host["services"])
-
-        # Streaming bulk export
-        async for aggregation in client.bulk_export_stream(queries):
-            print(aggregation.events[0].ip)
-
-asyncio.run(main())
-```
-
 ## Documentation
 
 Docstrings are used to document the library.
@@ -95,160 +53,4 @@ If you need commercial support, have a look at https://leakix.net/plans.
 
 ## Examples
 
-```python
-import decouple
-from leakix import Client
-from leakix.query import MustQuery, MustNotQuery, RawQuery
-from leakix.field import PluginField, CountryField, TimeField, Operator
-from leakix.plugin import Plugin
-from datetime import datetime, timedelta
-
-
-API_KEY = decouple.config("API_KEY")
-BASE_URL = decouple.config("LEAKIX_HOST", default=None)
-CLIENT = Client(api_key=API_KEY)
-
-
-def example_get_host_filter_plugin():
-    response = CLIENT.get_host(ipv4="33.33.33.33")
-    assert response.status_code() == 200
-
-
-def example_get_service_filter_plugin():
-    """
-    Filter by fields. In this example, we want to have the NTLM services.
-    A list of plugins can be found in leakix.plugin
-    """
-    query_http_ntlm = MustQuery(field=PluginField(Plugin.HttpNTLM))
-    response = CLIENT.get_service(queries=[query_http_ntlm])
-    assert response.status_code() == 200
-    # check we only get NTML related services
-    assert all((i.tags == ["ntlm"] for i in response.json()))
-
-
-def example_get_service_filter_plugin_with_pagination():
-    """
-    Filter by fields. In this example, we want to have the NTLM services.
-    A list of plugins can be found in leakix.plugin.
-    Ask for page 1 (starts at 0)
-    """
-    query_http_ntlm = MustQuery(field=PluginField(Plugin.HttpNTLM))
-    response = CLIENT.get_service(queries=[query_http_ntlm], page=1)
-    assert response.status_code() == 200
-    # check we only get NTML related services
-    assert all((i.tags == ["ntlm"] for i in response.json()))
-
-
-def example_get_leaks_filter_multiple_plugins():
-    query_http_ntlm = MustQuery(field=PluginField(Plugin.HttpNTLM))
-    query_country = MustQuery(field=CountryField("France"))
-    response = CLIENT.get_leak(queries=[query_http_ntlm, query_country])
-    assert response.status_code() == 200
-    assert all(
-        (
-            i.geoip.country_name == "France" and i.tags == ["ntlm"]
-            for i in response.json()
-        )
-    )
-
-
-def example_get_leaks_multiple_filter_plugins_must_not():
-    query_http_ntlm = MustQuery(field=PluginField(Plugin.HttpNTLM))
-    query_country = MustNotQuery(field=CountryField("France"))
-    response = CLIENT.get_leak(queries=[query_http_ntlm, query_country])
-    assert response.status_code() == 200
-    assert all(
-        (
-            i.geoip.country_name != "France" and i.tags == ["ntlm"]
-            for i in response.json()
-        )
-    )
-
-
-def example_get_leak_raw_query():
-    raw_query = '+plugin:HttpNTLM +country:"France"'
-    query = RawQuery(raw_query)
-    response = CLIENT.get_leak(queries=[query])
-    assert response.status_code() == 200
-    assert all(
-        (
-            i.geoip.country_name == "France" and i.tags == ["ntlm"]
-            for i in response.json()
-        )
-    )
-
-
-def example_get_leak_plugins_with_time():
-    query_plugin = MustQuery(field=PluginField(Plugin.GitConfigHttpPlugin))
-    today = datetime.now()
-    one_month_ago = today - timedelta(days=30)
-    query_today = MustQuery(field=TimeField(today, Operator.StrictlySmaller))
-    query_yesterday = MustQuery(
-        field=TimeField(one_month_ago, Operator.StrictlyGreater)
-    )
-    queries = [query_today, query_yesterday, query_plugin]
-    response = CLIENT.get_leak(queries=queries)
-    assert response.status_code() == 200
-
-
-def example_get_plugins():
-    response = CLIENT.get_plugins()
-    for p in response.json():
-        print(p.name)
-        print(p.description)
-
-
-def example_search_simple():
-    """
-    Simple search using query string syntax (same as the website).
-    No need to build Query objects manually.
-    """
-    response = CLIENT.search("+plugin:GitConfigHttpPlugin", scope="leak")
-    for event in response.json():
-        print(event.ip)
-
-
-def example_search_service():
-    """
-    Search for services with multiple filters.
-    """
-    response = CLIENT.search("+country:FR +port:22", scope="service")
-    for event in response.json():
-        print(event.ip, event.port)
-
-
-def example_get_domain():
-    """
-    Get services and leaks for a domain.
-    """
-    response = CLIENT.get_domain("example.com")
-    if response.is_success():
-        print("Services:", response.json()["services"])
-        print("Leaks:", response.json()["leaks"])
-
-
-def example_bulk_stream():
-    """
-    Streaming bulk export - memory efficient for large datasets.
-    Results are yielded one by one instead of loading all into memory.
-    """
-    query = MustQuery(field=PluginField(Plugin.GitConfigHttpPlugin))
-    for aggregation in CLIENT.bulk_export_stream(queries=[query]):
-        for event in aggregation.events:
-            print(event.ip)
-
-
-if __name__ == "__main__":
-    example_get_host_filter_plugin()
-    example_get_service_filter_plugin()
-    example_get_service_filter_plugin_with_pagination()
-    example_get_leaks_filter_multiple_plugins()
-    example_get_leaks_multiple_filter_plugins_must_not()
-    example_get_leak_plugins_with_time()
-    example_get_leak_raw_query()
-    example_get_plugins()
-    example_search_simple()
-    example_search_service()
-    example_get_domain()
-    example_bulk_stream()
-```
+See [example/](example/) for complete usage examples.

example/example_async_client.py

@@ -0,0 +1,43 @@
+import asyncio
+
+import decouple
+
+from leakix import AsyncClient
+
+API_KEY = decouple.config("API_KEY")
+
+
+async def example_search():
+    """Simple async search using query string syntax."""
+    async with AsyncClient(api_key=API_KEY) as client:
+        results = await client.search("+plugin:GitConfigHttpPlugin", scope="leak")
+        for event in results:
+            print(event.ip, event.host)
+
+
+async def example_get_host():
+    """Async host lookup."""
+    async with AsyncClient(api_key=API_KEY) as client:
+        host = await client.get_host("8.8.8.8")
+        print("Services:", host["services"])
+        print("Leaks:", host["leaks"])
+
+
+async def example_get_domain():
+    """Async domain lookup."""
+    async with AsyncClient(api_key=API_KEY) as client:
+        domain = await client.get_domain("example.com")
+        print("Services:", domain["services"])
+        print("Leaks:", domain["leaks"])
+
+
+async def example_bulk_export_stream():
+    """Async streaming bulk export - memory efficient for large datasets."""
+    async with AsyncClient(api_key=API_KEY) as client:
+        async for aggregation in client.bulk_export_stream("+plugin:GitConfigHttpPlugin"):
+            for event in aggregation["events"]:
+                print(event["ip"])
+
+
+if __name__ == "__main__":
+    asyncio.run(example_search())

example/example_client.py

@@ -118,6 +118,36 @@ def example_get_subdomains():
     print(response.json())
 
 
+def example_search_simple():
+    """Simple search using query string syntax (same as the website)."""
+    response = CLIENT.search("+plugin:GitConfigHttpPlugin", scope="leak")
+    for event in response.json():
+        print(event.ip)
+
+
+def example_search_service():
+    """Search for services with multiple filters."""
+    response = CLIENT.search("+country:FR +port:22", scope="service")
+    for event in response.json():
+        print(event.ip, event.port)
+
+
+def example_get_domain():
+    """Get services and leaks for a domain."""
+    response = CLIENT.get_domain("example.com")
+    if response.is_success():
+        print("Services:", response.json()["services"])
+        print("Leaks:", response.json()["leaks"])
+
+
+def example_bulk_export_stream():
+    """Streaming bulk export - memory efficient for large datasets."""
+    query = MustQuery(field=PluginField(Plugin.GitConfigHttpPlugin))
+    for aggregation in CLIENT.bulk_export_stream(queries=[query]):
+        for event in aggregation.events:
+            print(event.ip)
+
+
 if __name__ == "__main__":
     example_get_host_filter_plugin()
     example_get_service_filter_plugin()
@@ -131,3 +161,7 @@ def example_get_subdomains():
     example_bulk_service()
     example_bulk_export_last_event()
     example_get_subdomains()
+    example_search_simple()
+    example_search_service()
+    example_get_domain()
+    example_bulk_export_stream()

leakix/async_client.py

@@ -176,8 +176,9 @@ async def get_host(self, ip: str) -> dict:
         """
         status, data = await self._get(f"/host/{ip}")
         if status == 200 and isinstance(data, dict):
-            services = data.get("Services") or []
-            leaks = data.get("Leaks") or []
+            # Handle both old (uppercase) and new (lowercase) API response formats
+            services = data.get("services") or data.get("Services") or []
+            leaks = data.get("leaks") or data.get("Leaks") or []
             return {
                 "services": self._parse_events(services),
                 "leaks": self._parse_events(leaks),
@@ -206,8 +207,9 @@ async def get_domain(self, domain: str) -> dict:
         """
         status, data = await self._get(f"/domain/{domain}")
         if status == 200 and isinstance(data, dict):
-            services = data.get("Services") or []
-            leaks = data.get("Leaks") or []
+            # Handle both old (uppercase) and new (lowercase) API response formats
+            services = data.get("services") or data.get("Services") or []
+            leaks = data.get("leaks") or data.get("Leaks") or []
             return {
                 "services": self._parse_events(services),
                 "leaks": self._parse_events(leaks),

leakix/client.py

@@ -148,10 +148,12 @@ def get_host(self, ipv4: str) -> AbstractResponse:
         r = self.__get(url, params=None)
         if r.is_success():
             response_json = r.json()
-            formatted_result = HostResult.from_dict(response_json)
+            # Handle both old (uppercase) and new (lowercase) API response formats
+            services = response_json.get("services") or response_json.get("Services") or []
+            leaks = response_json.get("leaks") or response_json.get("Leaks") or []
             response_json = {
-                "services": formatted_result.Services,
-                "leaks": formatted_result.Leaks,
+                "services": services,
+                "leaks": leaks,
             }
             r.response_json = response_json
         return r
@@ -253,10 +255,12 @@ def get_domain(self, domain: str) -> AbstractResponse:
         r = self.__get(url, params=None)
         if r.is_success():
             response_json = r.json()
-            formatted_result = HostResult.from_dict(response_json)
+            # Handle both old (uppercase) and new (lowercase) API response formats
+            services = response_json.get("services") or response_json.get("Services") or []
+            leaks = response_json.get("leaks") or response_json.get("Leaks") or []
             response_json = {
-                "services": formatted_result.Services,
-                "leaks": formatted_result.Leaks,
+                "services": services,
+                "leaks": leaks,
             }
             r.response_json = response_json
         return r