LEXIO-37868 Initial CRMA REST API client (#1)

* init

* conn

* readme

* cci

* Version bumped to 0.2.0

* interface

* readme

Co-authored-by: ns-circle-ci <[email protected]>

Author: jdrake

.circleci/config.yml

@@ -8,6 +8,7 @@
 **/*.egg-info
 **/*.pyc
 **/dist
+not a tty
 
 # Pycharm
 .idea

.gitignore

@@ -67,7 +67,7 @@ repos:
         entry: black
         language: python
         types: [file, python]
-        additional_dependencies: [black==21.9b0]
+        additional_dependencies: [black==22.3.0]
 
       - id: mypy
         name: Type check Python (mypy)

.pre-commit-config.yaml

@@ -5,7 +5,8 @@ CRM Analytics REST API Client
 
 Features:
 
-- <!-- list of features -->
+- Execute SAQL queries
+- List dataset versions
 
 Table of Contents:
 
@@ -25,7 +26,46 @@ poetry add crma-api-client
 
 ## Guide
 
-<!-- Subsections explaining how to use the package -->
+First, you need to create a new client instance. To do that, you either need to have credentials for an OAuth app or an existing access token handy:
+
+```python
+from crma_api_client.client import ConnectionInfo, CRMAAPIClient
+
+# Generate connection info if you don't already have an access token
+conn = await ConnectionInfo.generate(
+    client_id="abc123",
+    client_secret="***",
+    username="[email protected]",
+    password="***"
+)
+
+# If you already have an instance URL and access token, you can instantiate directly
+conn = ConnectionInfo(instance_url="https://company.my.salesforce.com", access_token="XYZ123")
+
+# Create the client, passing in the connection object
+client = CRMAAPIClient(conn)
+```
+
+Next, you can use methods on the client to make requests:
+
+```python
+response = await client.list_dataset_versions("Sample_Superstore_xls_Orders")
+version = response.versions[0]
+query = "\n".join(
+    [
+        f"""q = load "{version.dataset.id}/{version.id}";""",
+        """q = group q by 'Category';""",
+        """q = foreach q generate q.'Category' as 'Category', sum(q.'Sales') as 'Sales';""",
+        """q = order q by 'Category' asc;""",
+    ]
+)
+response = await client.query(query)
+assert response.results.records == [
+    {"Category": "Furniture", "Sales": 741999.7953},
+    {"Category": "Office Supplies", "Sales": 719047.032},
+    {"Category": "Technology", "Sales": 836154.033},
+]
+```
 
 ## Development
 

README.md

@@ -1,3 +1,5 @@
 """CRM Analytics REST API Client"""
 
-__version__ = "0.1.0"
+from .client import CRMAAPIClient
+
+__version__ = "0.2.0"

crma_api_client/__init__.py

@@ -0,0 +1,201 @@
+"""Contains the CRMA API client"""
+
+import logging
+from typing import Any, Dict, Optional
+from uuid import uuid4
+
+import backoff
+import httpx
+from pydantic import BaseModel
+
+from crma_api_client.resources.dataset import DatasetVersionsResponse
+from crma_api_client.resources.query import QueryLanguage, QueryResponse
+from .encoder import json_dumps_common
+
+logger = logging.getLogger(__name__)
+
+
+class ConnectionInfo(BaseModel):
+    """Model with info for making API requests to a Salesforce instance"""
+
+    instance_url: str
+    access_token: str
+    token_type: str = "Bearer"
+
+    @property
+    def authorization(self) -> str:
+        """Returns authorization header value"""
+        return f"{self.token_type} {self.access_token}"
+
+    @classmethod
+    async def generate(
+        cls,
+        client_id: str,
+        client_secret: str,
+        username: str,
+        password: str,
+        grant_type: str = "password",
+    ) -> "ConnectionInfo":
+        """Create a connection info object by generating a fresh token
+
+        See https://developer.salesforce.com/docs/atlas.en-us.bi_dev_guide_rest.meta/bi_dev_guide_rest/bi_rest_authentication.htm
+
+        Args:
+            client_id: OAuth app client ID
+            client_secret: OAuth app client secret
+            username: Username for the user calling the API
+            password: Password for the user calling the API
+            grant_type: OAuth grant type
+
+        Returns:
+            new ConnectionInfo object
+
+        """
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                "https://login.salesforce.com/services/oauth2/token",
+                data={
+                    "client_id": client_id,
+                    "client_secret": client_secret,
+                    "username": username,
+                    "password": password,
+                    "grant_type": grant_type,
+                },
+                headers={"Accept": "application/json"},
+            )
+            return cls.parse_obj(response.json())
+
+
+class CRMAAPIClient:
+    """CRM Analytics REST API client"""
+
+    def __init__(
+        self,
+        conn: ConnectionInfo,
+        version: str = "v54.0",
+        timeout: float = 60.0,
+        connect_timeout: float = 5.0,
+        logger: logging.Logger = logger,
+    ) -> None:
+        """Initialize the CRMAAPIClient
+
+        Args:
+            conn: Object containing for making API requests to a Salesforce instance.
+            version: CRMA REST API version
+            timeout: Default timeout for requests and non-connect operations, in seconds
+            connect_timeout: Default timeout for establishing an HTTP connection, in
+                seconds
+            logger: Custom logger instance to use instead of the stdlib
+
+        """
+        self.logger = logger
+        self._client = httpx.AsyncClient(
+            base_url=conn.instance_url.rstrip("/") + f"/services/data/{version}",
+            headers={"Authorization": conn.authorization},
+            timeout=httpx.Timeout(timeout, connect=connect_timeout),
+        )
+
+    async def _get_headers(
+        self,
+    ) -> Dict[str, str]:
+        """Get a headers dict from the calling context
+
+        Returns:
+            Dict containing headers to include in requests
+
+        """
+        headers = {
+            "content-type": "application/json",
+            "accept": "application/json",
+        }
+
+        return headers
+
+    @backoff.on_exception(
+        backoff.expo,
+        httpx.HTTPStatusError,
+        max_tries=3,
+        giveup=lambda e: e.response.status_code != 502,
+    )
+    async def request(
+        self,
+        path: str,
+        method: str,
+        json_data: Optional[Any] = None,
+        params: Optional[Dict[str, Any]] = None,
+        **kwargs: Any,
+    ) -> httpx.Response:
+        """Generic method to send a JSON request to the service
+
+        Args:
+            path: Path to the API resource. This path will be appended to the base url.
+            method: HTTP method
+            json_data: Request payload (for POST/PUT/PATCH requests)
+            params: Request query params
+
+        Returns:
+            response object
+
+        """
+        path = "/" + path.strip("/")
+        if json_data:
+            json_data = json_dumps_common(json_data).encode()
+        headers = await self._get_headers()
+        self.logger.debug(f"Service request starting path={path} method={method}")
+        response = await self._client.request(
+            method.upper(),
+            path,
+            headers=headers,
+            content=json_data,
+            params=params,
+            **kwargs,
+        )
+        self.logger.debug(
+            f"Service request completed status_code={response.status_code}"
+        )
+        response.raise_for_status()
+        return response
+
+    async def list_dataset_versions(self, identifier: str) -> DatasetVersionsResponse:
+        """List the versions for a dataset
+
+        Args:
+            identifier: Dataset name or ID
+
+        Returns:
+            DatasetVersionsResponse object
+
+        """
+        response = await self.request(f"/wave/datasets/{identifier}/versions", "GET")
+        return DatasetVersionsResponse.parse_obj(response.json())
+
+    async def query(
+        self,
+        query: str,
+        query_language: QueryLanguage = QueryLanguage.saql,
+        name: Optional[str] = None,
+        timezone: Optional[str] = None,
+    ) -> QueryResponse:
+        """Execute a query
+
+        Args:
+            query: Query string
+            query_language: Query language. One of: SAQL (default), SQL
+            name: Query name. Defaults to a UUID
+            timezone: Timezone for the query
+
+        Returns:
+            QueryResponse object
+
+        """
+        json_data = {
+            "query": query,
+            "name": name or str(uuid4()),
+            "queryLanguage": query_language.value,
+        }
+        if timezone:
+            json_data["timezone"] = timezone
+
+        response = await self.request("/wave/query", "POST", json_data=json_data)
+
+        return QueryResponse.parse_obj(response.json())

crma_api_client/client.py

@@ -0,0 +1,69 @@
+"""Contains JSON encoders"""
+
+from datetime import date, datetime
+from enum import Enum
+import functools
+import json
+from typing import Any
+
+
+class CommonEncoder(json.JSONEncoder):
+    """Custom JSON encoder to serialize commonly-used objects"""
+
+    def _encode(self, obj: Any) -> Any:
+        """Method to handle serialization of custom objects
+
+        Defaults to stringifying the object.
+
+        Args:
+            obj: Node in the data structure to serialize
+
+        Returns:
+            transformed data structure to include in final JSON output
+
+        """
+        if isinstance(obj, (str, bool, int, float)):
+            return obj
+        elif isinstance(obj, bytes):
+            return obj.decode()
+        elif isinstance(obj, datetime):
+            return obj.isoformat()
+        elif isinstance(obj, date):
+            return str(obj)
+        elif isinstance(obj, set):
+            return {self._encode(key): True for key in obj}
+        elif isinstance(obj, Enum):
+            return obj.value
+
+        for method in ("to_dict", "dict"):
+            to_dict_method = getattr(obj, method, None)
+            if callable(to_dict_method):
+                obj = to_dict_method()
+                break
+
+        if isinstance(obj, dict):
+            return {
+                self._encode(key): self._encode(value) for key, value in obj.items()
+            }
+        elif isinstance(obj, (list, tuple)):
+            return [self._encode(value) for value in obj]
+
+        return str(obj)
+
+    def encode(self, obj: Any) -> str:
+        """Method to handle serialization of objects
+
+        Proxies to custom internal method.
+
+        Args:
+            obj: Node in the data structure to serialize
+
+        Returns:
+            serialized JSON output
+
+        """
+        return super().encode(self._encode(obj))
+
+
+#: Function to serialize a data structure using the CommonEncoder
+json_dumps_common = functools.partial(json.dumps, cls=CommonEncoder)