feat(ad-hoc): compose your requests

alice/onboarding/models/ad_hoc_executor.py

@@ -0,0 +1,52 @@
+from dataclasses import dataclass
+from typing import Any, Union
+
+from requests import Response, Session
+
+
+@dataclass
+class AdHocExecutor:
+    session: Session
+    headers: dict[str, Any]
+    base_url: str
+    timeout: float
+
+    def get_with(self, url_path: str) -> Response:
+        return self.session.get(
+            f"{self.base_url}/{url_path}", headers=self.headers, timeout=self.timeout
+        )
+
+    def post_with(
+        self, url_path: str, json: Union[dict[str, Any], None] = None
+    ) -> Response:
+        return self.session.post(
+            f"{self.base_url}/{url_path}",
+            json=json,
+            headers=self.headers,
+            timeout=self.timeout,
+        )
+
+    def patch_with(
+        self, url_path: str, json: Union[dict[str, Any], None] = None
+    ) -> Response:
+        return self.session.patch(
+            f"{self.base_url}/{url_path}",
+            json=json,
+            headers=self.headers,
+            timeout=self.timeout,
+        )
+
+    def put_with(
+        self, url_path: str, json: Union[dict[str, Any], None] = None
+    ) -> Response:
+        return self.session.put(
+            f"{self.base_url}/{url_path}",
+            json=json,
+            headers=self.headers,
+            timeout=self.timeout,
+        )
+
+    def delete_with(self, url_path: str) -> Response:
+        return self.session.delete(
+            f"{self.base_url}/{url_path}", headers=self.headers, timeout=self.timeout
+        )

alice/onboarding/onboarding.py

@@ -1,8 +1,8 @@
 from datetime import datetime
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Callable, Dict, List, Optional, Union
 
 from meiga import Failure, Result, Success, early_return, isSuccess
-from requests import Session
+from requests import Response, Session
 
 from alice.auth.auth import Auth
 from alice.auth.auth_errors import AuthError
@@ -17,6 +17,7 @@
 from alice.onboarding.enums.onboarding_steps import OnboardingSteps
 from alice.onboarding.enums.user_state import UserState
 from alice.onboarding.enums.version import Version
+from alice.onboarding.models.ad_hoc_executor import AdHocExecutor
 from alice.onboarding.models.bounding_box import BoundingBox
 from alice.onboarding.models.device_info import DeviceInfo
 from alice.onboarding.models.report.report import Report
@@ -2018,3 +2019,23 @@ def update_user_flow(
                     operation="update_user_flow", response=response
                 )
             )
+
+    @early_return
+    def ad_hoc(
+        self,
+        func: Callable[[AdHocExecutor], Response],
+        user_id: Union[str, None] = None,
+        verbose: bool = False,
+    ) -> Result[bool, OnboardingError]:
+        response = self.onboarding_client.ad_hoc(
+            func=func,
+            user_id=user_id,
+            verbose=verbose,
+        ).unwrap_or_return()
+
+        if response.status_code in range(200, 300):
+            return Success(response)
+        else:
+            return Failure(
+                OnboardingError.from_response(operation="ad_hoc", response=response)
+            )

alice/onboarding/onboarding_client.py

@@ -1,7 +1,7 @@
 import json
 import platform
 from datetime import datetime
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Callable, Dict, List, Optional, Union
 
 import requests
 from meiga import Error, Failure, Result, Success, early_return
@@ -19,6 +19,7 @@
 from alice.onboarding.enums.onboarding_steps import OnboardingSteps
 from alice.onboarding.enums.user_state import UserState
 from alice.onboarding.enums.version import Version
+from alice.onboarding.models.ad_hoc_executor import AdHocExecutor
 from alice.onboarding.models.bounding_box import BoundingBox
 from alice.onboarding.models.device_info import DeviceInfo
 from alice.onboarding.models.user_info import UserInfo
@@ -2332,3 +2333,35 @@ def update_user_flow(
         print_response(response=response, verbose=verbose)
 
         return Success(response)
+
+    @early_return
+    @timeit
+    def ad_hoc(
+        self,
+        func: Callable[[AdHocExecutor], Response],
+        user_id: Union[str, None] = None,
+        verbose: bool = False,
+    ) -> Result[Response, Error]:
+
+        print_intro("ad_hoc", verbose=verbose)
+
+        token = self.auth.create_backend_token(user_id).unwrap_or_return()
+        print_token("backend_token", token, verbose=verbose)
+
+        headers = self._auth_headers(token)
+
+        try:
+            response = func(
+                AdHocExecutor(
+                    session=self.session,
+                    headers=headers,
+                    base_url=self.url,
+                    timeout=self.timeout,
+                )
+            )
+
+        except requests.exceptions.Timeout:
+            return Failure(OnboardingError.timeout(operation="ad_hoc"))
+        print_response(response=response, verbose=verbose)
+
+        return Success(response)

examples/onboarding_ad_hoc.py

@@ -0,0 +1,38 @@
+import os
+from typing import Optional
+
+from requests import Response
+
+from alice import Config, Onboarding
+from alice.onboarding.models.ad_hoc_executor import AdHocExecutor
+
+
+def onboarding_ad_hoc(api_key: str, verbose: Optional[bool] = False) -> None:
+    config = Config(api_key=api_key, verbose=verbose)
+    onboarding = Onboarding.from_config(config)
+
+    # Create User (Backend without user_id)
+    def create_user(executor: AdHocExecutor) -> Response:
+        return executor.post_with(url_path="user")
+
+    response = onboarding.ad_hoc(create_user).unwrap_or_raise()
+    user_id = response.json().get("user_id")
+    print(f"{user_id=}")
+
+    # Get User State (Backend with user_id)
+    def get_user_state(executor: AdHocExecutor) -> Response:
+        return executor.get_with(url_path="/user/state")
+
+    response = onboarding.ad_hoc(get_user_state, user_id=user_id).unwrap_or_raise()
+    state = response.json()
+    print(f"{state=}")
+
+
+if __name__ == "__main__":
+    api_key = os.environ.get("ONBOARDING_API_KEY")
+    if api_key is None:
+        raise AssertionError(
+            "Please configure your ONBOARDING_API_KEY to run the example"
+        )
+    print("Running onboarding example...")
+    onboarding_ad_hoc(api_key=api_key, verbose=True)