pycarol.carol

class pycarol.carol.Carol(domain=None, app_name=None, auth=None, connector_id=None, port=443, verbose=False, organization=None, environment=None, host=None)[source]

This class handle all Carol`s API calls It will handle all API calls, for a given authentication method. :param domain: str.

Args:

domain: str. default None.
Tenant name. e.x., domain.carol.ai
app_name: str. default None.
Carol app name.
auth: PwdAuth or ApiKeyAuth.
object Auth Carol object to handle authentication
connector_id: str , default __CONNECTOR_PYCAROL__.
Connector Id
port: int , default 443.
Port to be used (when running locally it could change)
verbose: bool , default False.
If True will print the header, method and URL of each API call.
organization: str , default None.
Organization domain.
environment: str, default carol.ai,

Which Carol’s environment to use. There are three possible values today.

  1. ‘carol.ai’ for the production environment
  2. ‘karol.ai’ for the explore environment
  3. ‘qarol.ai’ for the QA environment
host: str default None

This will overwrite the host used. Today the host is:

  1. if organization is None, host={domain}.{environment}
  2. else host={organization}.{environment}

See Carol._set_host.

OBS:

In case all parameters are None, pycarol will try yo find their values in the environment variables. The values are:

  1. CAROLTENANT for domain
  2. CAROLAPPNAME for app_name
  3. CAROLAPPOAUTH for auth
  4. CAROLORGANIZATION for organization
  5. CAROLCONNECTORID for connector_id
  6. CAROL_DOMAIN for environment
  7. CAROLUSER for carol user email
  8. CAROLPWD for user password.
api_key_details(api_key, connector_id)[source]

Display information about the API key.

Args:

api_key: str
Carol’s api key
connector_id: str
Connector Id which API key was created.
Returns: dict
Dictionary with API key information.
api_key_revoke(connector_id)[source]

Revoke API key for ta given connector_id

Args:

connector_id: str
Connector Id which API key was created.

Returns: dict

Dictionary with API request response.
call_api(path, method=None, data=None, auth=True, params=None, content_type='application/json', retries=8, session=None, backoff_factor=0.5, status_forcelist=(502, 503, 504, 524), downloadable=False, method_whitelist=frozenset({'PUT', 'HEAD', 'OPTIONS', 'GET', 'DELETE', 'TRACE'}), errors='raise', extra_headers=None, **kwds)[source]

This method handles all the API calls.

Args:

path: str.
API URI path. e.x. v2/staging/schema
method: ‘str’, default None.
Set of uppercased HTTP method verbs that we should call on.
data: ‘dict`, default None.
Dictionary, list of tuples, bytes, or file-like object to send in the body of the request.
auth: :class: pycarol.ApiKeyAuth or pycarol.PwdAuth
Auth type to be used within the API’s calls.
params: (optional) Dictionary, list of tuples or bytes to send
in the query string for the requests.Request.
content_type: str, default ‘application/json’
Content type for the api call
retries: int , default 5
Number of retries for the API calls
session: :class requests.Session object dealt None
It allows you to persist certain parameters across requests.
backoff_factor: float , default 0.5
Backoff factor to apply between attempts. It will sleep for:
{backoff factor} * (2 ^ ({retries} - 1)) seconds
status_forcelist: iterable , default (500, 502, 503, 504, 524).
A set of integer HTTP status codes that we should force a retry on. A retry is initiated if the request method is in method_whitelist and the response status code is in status_forcelist.
downloadable: bool default False.
If the request will return a file to download.
method_whitelist: iterable , default frozenset([‘HEAD’, ‘TRACE’, ‘GET’, ‘PUT’, ‘OPTIONS’, ‘DELETE’]))
Set of uppercased HTTP method verbs that we should retry on.
errors: {‘ignore’, ‘raise’}, default ‘raise’
If ‘raise’, then invalid request will raise an exception If ‘ignore’, then invalid request will return the request response
extra_headers: dict default None
extra headers to be sent.
kwds: dict default None
Extra parameters to be sent to :class: requests.request
Rerturn:
Dict with API response.
copy_token()[source]

Copy token to clipboard

Returns:
None
issue_api_key()[source]

Create an API key for a given connector.

Returns: dict
Dictionary with the API key.
switch_context(env_name=None, env_id=None, app_name=None)[source]

Context manager to temporary have access to a second environment

Args:

env_name: str default None
Environment (tenant) name to switch the context to.
env_id: str default None
Environment (tenant) id to switch the context to.
app_name: str default None
App name in the target environment to switch the context to. Only needed with using CDS.
Returns:
None

Usage:

from pycarol import Carol, Staging
carol = Carol('B', 'teste', auth=PwdAuth('email@totvs.com.br', 'pass'), )
with carol.switch_context('A') as carol_tenant_A:
    Staging(carol_tenant_A).fetch_parquet(...) # fetch parquet from tenant A
#back to tenant B
switch_environment(env_name=None, env_id=None, app_name=None)[source]

Switch environments. If the user has access to this environment, it will be “logged in” in this new environment.

Args:

env_name: str default None
Environment (tenant) name to switch the context to.
env_id: str default None
Environment (tenant) id to switch the context to.
app_name: str default None
App name in the target environment to switch the context to. Only needed with using CDS.
Returns:
self
from pycarol import Carol, Staging
carol = Carol('B', 'teste', auth=PwdAuth('email@totvs.com.br', 'pass'), )
carol.switch_environment('A')
Staging(carol_tenant_A).fetch_parquet(...) # fetch parquet from tenant A
# To switch back
carol.switch_environment('B')
#back to tenant B