feat(event): Add support for ingest API

Author: vincent-pochet

lago_python_client/base_client.py

@@ -7,9 +7,10 @@
 class BaseClient(ABC):
     """The base class used for each collection client."""
 
-    def __init__(self, base_url: str, api_key: str):
+    def __init__(self, base_url: str, api_key: str, base_ingest_url: str = ""):
         self.base_url = base_url
         self.api_key = api_key
+        self.base_ingest_url = base_ingest_url
 
     @property
     @classmethod

lago_python_client/client.py

@@ -31,16 +31,32 @@
 
 class Client:
     BASE_URL: Final[str] = "https://api.getlago.com/"
+    BASE_INGEST_URL: Final[str] = "https://ingest.getlago.com/"
     API_PATH: Final[str] = "api/v1/"
 
-    def __init__(self, api_key: str = "", api_url: str = "") -> None:
+    def __init__(self, api_key: str = "", api_url: str = "", use_ingest_service = False, ingest_api_url: str = "") -> None:
         self.api_key: str = api_key
         self.api_url: str = api_url
+        self.use_ingest_service: bool = use_ingest_service
+        self.ingest_api_url: str = ingest_api_url
 
     @property
     def base_api_url(self) -> str:
         return urljoin(self.api_url if self.api_url else Client.BASE_URL, Client.API_PATH)
 
+    @property
+    def base_ingest_api_url(self) -> str:
+        if not self.use_ingest_service:
+            return self.base_api_url
+
+        if self.ingest_api_url == "":
+            ingest_url = Client.BASE_INGEST_URL
+        else:
+            ingest_url = self.ingest_api_url
+
+        return urljoin(ingest_url, Client.API_PATH)
+
+
     @callable_cached_property
     def add_ons(self) -> AddOnClient:
         return AddOnClient(self.base_api_url, self.api_key)
@@ -67,7 +83,7 @@ def customers(self) -> CustomerClient:
 
     @callable_cached_property
     def events(self) -> EventClient:
-        return EventClient(self.base_api_url, self.api_key)
+        return EventClient(self.base_api_url, self.api_key, self.base_ingest_api_url)
 
     @callable_cached_property
     def fees(self) -> FeeClient:

lago_python_client/events/clients.py

@@ -1,17 +1,20 @@
 import sys
-from typing import Any, ClassVar, Type
+from typing import Any, ClassVar, Optional, Type
+
+import httpx
 
 from lago_python_client.base_model import BaseModel
 
 from ..base_client import BaseClient
 from ..fees.clients import FeeClient
-from ..mixins import CreateCommandMixin, FindCommandMixin
+from ..mixins import FindCommandMixin
 from ..models.event import EventResponse
 from ..models.fee import FeeResponse
 from ..services.json import to_json
 from ..services.request import make_headers, make_url, send_post_request
 from ..services.response import (
     get_response_data,
+    prepare_object_response,
     prepare_object_list_response,
     verify_response,
     Response,
@@ -23,11 +26,34 @@
     from typing import Mapping
 
 
-class EventClient(CreateCommandMixin[EventResponse], FindCommandMixin[EventResponse], BaseClient):
+class EventClient(FindCommandMixin[EventResponse], BaseClient):
     API_RESOURCE: ClassVar[str] = "events"
     RESPONSE_MODEL: ClassVar[Type[EventResponse]] = EventResponse
     ROOT_NAME: ClassVar[str] = "event"
 
+    def create(self, input_object: BaseModel, timeout: Optional[httpx.Timeout] = None) -> Optional[EventResponse]:
+        api_response: Response = send_post_request(
+            url=make_url(
+                origin=self.base_ingest_url,
+                path_parts=(self.API_RESOURCE,),
+            ),
+            content=to_json({
+                self.ROOT_NAME: input_object.dict(),
+            }),
+            headers=make_headers(api_key=self.api_key),
+            timeout=timeout,
+        )
+
+        # Process response data
+        response_data = get_response_data(response=api_response, key=self.ROOT_NAME)
+        if not response_data:
+            return None
+
+        return prepare_object_response(
+            response_model=self.RESPONSE_MODEL,
+            data=response_data,
+        )
+
     def batch_create(self, input_object: BaseModel) -> None:
         api_response: Response = send_post_request(
             url=make_url(

tests/test_event_client.py

@@ -80,6 +80,12 @@ def test_invalid_create_events_request(httpx_mock: HTTPXMock):
     with pytest.raises(LagoApiError):
         client.events.create(create_event())
 
+def test_valid_create_event_request_with_custom_ingest_url(httpx_mock: HTTPXMock):
+    client = Client(api_key='886fe239-927d-4072-ab72-6dd345e8dd0d', use_ingest_service=True, ingest_api_url='https://custom-ingest.getlago.com')
+
+    httpx_mock.add_response(method='POST', url='https://custom-ingest.getlago.com/api/v1/events', content=b'')
+    client.events.create(create_event())  # Any response means success, any exception - failure
+
 
 def test_valid_create_batch_events_request(httpx_mock: HTTPXMock):
     client = Client(api_key="886fe239-927d-4072-ab72-6dd345e8dd0d")