Mopinion Client

The intention of developing a MopinionClient is to make easy, beautiful and elegant when interacting with our API.

Credentials can be created via the Mopinion Suite at Integrations » Feedback API in classic interface or in the Raspberry interface, provided your package includes API access.

You can also take a look at this link with the steps to get private_key and public_key

MopinionClient Specifications

class mopinion_client.MopinionClient(public_key: str, private_key: str, max_retries: int = 3)

Client to interact with Mopinion API.

Provides functionality for authentication, authorization and requesting resources. Steps during instantiation:

  1. Credential validations.

  2. Instantiation of a session object from requests.Session that will be used in each request.

  3. Retrieval of signature_token from the API for an specific private_key and public_key.

When instantiating, a signature token is retrieved from the API and stored in the signature_token attribute using your private_key and public_key. The signature_token will be used in each request.

In each request, a HMAC signature will be created using SHA256-hashing, and encrypted with your signature_token. This HMAC signature is encode together with the public_key. After this encryption, the xtoken is set into the headers under the X-Auth-Token key.

Parameters
  • public_key (str) –

  • private_key (str) –

  • max_retries (int) – Defaults to 3.

is_available(verbose: bool = False) → Union[dict, bool]

Test the API’s availability.

It return a boolean True/False in case the API is available or not. In case we need extra information about the state of the API, we can provide a flag verbose=True.

Examples

>>> from mopinion_client import MopinionClient
>>> client = MopinionClient(public_key=PUBLICKEY, private_key=PRIVATEKEY)
>>> assert client.is_available()
>>> r = client.is_available(verbose=True)
>>> assert r["code"] == 200 and r["response"] == "pong" and r["version"] == "2.0.0"
request(endpoint: str, method: str = 'GET', version: str = '2.0.0', verbosity: str = 'normal', content_negotiation: str = 'application/json', body: Optional[dict] = None, query_params: Optional[dict] = None) → requests.models.Response

Generic method to send requests to our API.

Wrapper on top of requests.Session.request method adding token encryption on headers. Everytime we call request five steps are applied:

  1. Validation of arguments.

  2. Token creation - token depends on endpoint argument and signature_token.

  3. Preparation of parameter dictionary. Add token to headers.

  4. Make request.

  5. Return response.

Parameters
  • endpoint (str) – API endpoint.

  • method (str) – HTTP Method.

  • version (str) – API Version.

  • verbosity (str) – normal, quiet or full.

  • content_negotiation (str) – application/json or application/x-yaml.

  • body (dict) – Optional.

  • query_params (dict) – Optional.

Returns

response (requests.models.Response).

Examples

>>> from mopinion_client import MopinionClient
>>> client = MopinionClient(public_key=PUBLICKEY, private_key=PRIVATEKEY)
>>> response = client.request("/account")
>>> assert response.json()["_meta"]["code"] == 200
>>> response = client.request(endpoint="/deployments")
>>> assert response.json()["_meta"]["code"] == 200
>>> body = {"key": "key", "name": "My Test Deployment"},
>>> response = client.request(endpoint="/deployments", method="POST", body=body)
>>> assert response.json()["_meta"]["code"] == 201
>>> endpoint = "/deployments/abt34")
>>> response = client.request(endpoint, method="DELETE")
>>> assert response.json()["_meta"]["code"] == 200
resource(resource_name: str, resource_id: Optional[Union[str, int]] = None, sub_resource_name: Optional[str] = None, sub_resource_id: Optional[Union[str, int]] = None, method: str = 'GET', version: str = '2.0.0', verbosity: str = 'normal', content_negotiation: str = 'application/json', query_params: Optional[dict] = None, body: Optional[dict] = None, iterator: bool = False) → Union[requests.models.Response, collections.abc.Iterator]

Method to send requests to our API.

Abstraction of mopinion_api.MopinionClient.request. Interacts with the API in term of resources and subresources, and also, enables iterator protocol when requesting large resources.

Parameters
  • resource_name (str) –

  • resource_id (str/int) –

  • sub_resource_name (str) –

  • sub_resource_id (str) –

  • method (str) – HTTP Method.

  • version (str) – API Version.

  • verbosity (str) – normal, quiet or full.

  • content_negotiation (str) – application/json or application/x-yaml.

  • body (dict) – Optional.

  • query_params (dict) – Optional.

  • iterator (bool) – If sets to True an iterator will be returned.

Returns

response (requests.models.Response) or iterator (collections.abc.Iterator)

The endpoint is built from mopinion_api.dataclasses.ResourceUri and the parameters are:
  • resource_name (str) Required

  • resource_id (int/str) Optional

  • subresource_name (str) Optional

  • subresource_id (str) Optional

Resources and sub-resources options:
  • The resource_name options are: “account”, “deployments”, “datasets”, “reports”.

  • The subresource_name options are: “fields”, “feedback”.

You can also use the constants defined in the mopinion_api.MopinionClient class.
  • The resource_name options are: RESOURCE_ACCOUNT, RESOURCE_DEPLOYMENTS, RESOURCE_DATASETS, RESOURCE_REPORTS.

  • The subresource_name options are: SUBRESOURCE_FIELDS, SUBRESOURCE_FEEDBACK.

Examples

>>> from mopinion_client import MopinionClient
>>> client = MopinionClient(public_key=PUBLICKEY, private_key=PRIVATEKEY)
>>> response = client.resource("account")
>>> assert response.json()["_meta"]["code"] == 200
>>> response = client.resource(resource_name=client.RESOURCE_ACCOUNT)  # same as above
>>> assert response.json()["_meta"]["code"] == 200

When working with the API there is a limit of elements retrieved. The limit parameters defaults to 10. You can increase the limit, or you can request resources using the flag generator=True. This returns a Generator which traverses these pages for you and yields each result in the current page before retrieving the next page.

Examples

>>> from mopinion_client import MopinionClient
>>> client = MopinionClient(public_key=PUBLICKEY, private_key=PRIVATEKEY)
>>> iterator = client.resource("account", iterator=True)
>>> response = next(iterator)
>>> assert response.json()["_meta"]["code"] == 200

Below some more examples.

Examples

>>> from mopinion_client import MopinionClient
>>> client = MopinionClient(public_key=PUBLICKEY, private_key=PRIVATEKEY)
>>> response = client.resource("account")
>>> assert response.json()["_meta"]["code"] == 200
>>> response = client.resource("deployments")
>>> assert response.json()["_meta"]["code"] == 200
>>> body={"key": "mydeploymentkey3", "name": "My Test Deployment"},
>>> response = client.resource("deployments", method="POST", body=body)
>>> assert response.json()["_meta"]["code"] == 201
>>> response = client.resource("deployments", resource_id="abt34", method="DELETE")
>>> assert response.json()["_meta"]["code"] == 200