pycarlo 0.15.47

Monte Carlo's Python SDK

Pycarlo - Monte Carlo's Python SDK

Installation

Requires Python 3.10 or greater. Normally you can install and update using pip. For instance:

virtualenv venv
. venv/bin/activate

pip install -U pycarlo

Overview

Pycarlo comprises two components: core and features.

All Monte Carlo API queries and mutations that you could execute via the API are supported via the core library. Operations can be executed as first class objects, using sgqlc, or as raw GQL with variables. In both cases, a consistent object where fields can be referenced by dot notation and the more pythonic snake_case is returned for ease of use.

The features library provides additional convenience for performing common operations like with dbt, circuit breaking, and pii filtering.

Note that credentials are required to use the SDK — either an API key or OAuth client credentials. See our docs on API authentication for details.

Basic usage

Core

from pycarlo.core import Client, Query, Mutation

# First create a client. This creates a session using the 'default' profile from
# '~/.mcd/profiles.ini'. This profile is created automatically via
# `montecarlo configure` on the CLI. See the session subsection for
# customizations, options and alternatives (e.g. using the environment, params,
# named profiles, etc.)
client = Client()

# Now you can can execute a query. For instance, getUser (selecting the email field).
# This would be like executing -
#     curl --location --request POST 'https://api.getmontecarlo.com/graphql' \
#     --header 'x-mcd-id: <ID>' \
#     --header 'x-mcd-token: <TOKEN>' \
#     --header 'Content-Type: application/json' \
#     --data-raw '{"query": "query {getUser {email}}"}'
# Notice how the CamelCase from the Graphql query is converted to snake_case in
# both the request and response.
query = Query()
query.get_user.__fields__('email')
print(client(query).get_user.email)

# You can also execute a query that requires variables. For instance,
# testTelnetConnection (selecting all fields).
query = Query()
query.test_telnet_connection(host='montecarlodata.com', port=443)
print(client(query))

# If necessary, you can always generate (e.g. print) the raw query that would be executed.
print(query)
# query {
#   testTelnetConnection(host: "montecarlodata.com", port: 443) {
#     success
#     validations {
#       type
#       message
#     }
#     warnings {
#       type
#       message
#     }
#   }
# }

# If you are not a fan of sgqlc operations (Query and Mutation) you can also execute any
# raw query using the client. For instance, if we want the first 10 tables from getTables.
get_table_query = """
query getTables{
  getTables(first: 10) {
    edges {
      node {
        fullTableId
      }
    }
  }
}
"""
response = client(get_table_query)
# This returns a Box object where fields can be accessed using dot notation.
# Notice how unlike with the API the response uses the more Pythonic snake_case.
for edge in response.get_tables.edges:
    print(edge.node.full_table_id)
# The response can still be processed as a standard dictionary.
print(response['get_tables']['edges'][0]['node']['full_table_id'])

# You can also execute any mutations too. For instance, generateCollectorTemplate
# (selecting the templateLaunchUrl).
mutation = Mutation()
mutation.generate_collector_template().dc.template_launch_url()
print(client(mutation))

# Any errors will raise a GqlError with details. For instance, executing above with an
# invalid region.
mutation = Mutation()
mutation.generate_collector_template(region='artemis')
print(client(mutation))
# pycarlo.common.errors.GqlError: [
#   {'message': 'Region "\'artemis\'" not currently active.'...
# ]

Examples

Circuit Breaker Example

from pycarlo.core import Client, Session
from pycarlo.features.circuit_breakers import CircuitBreakerService

# Example from our test.snowflake account.
endpoint = "https://api.dev.getmontecarlo.com/graphql"

service = CircuitBreakerService(
    mc_client=Client(Session(mcd_profile="test-snow", endpoint=endpoint)), print_func=print
)
in_breach = service.trigger_and_poll(rule_uuid="87872875-fe80-4963-8ab0-c04397a6daae")
print("That can't be good. Our warehouse is broken." if in_breach else "Go, go, go!.")

Insight Upload Example

from pathlib import Path

import boto3
import requests

from pycarlo.core import Client, Query

MC_CLIENT = Client()
S3_CLIENT = boto3.client("s3")


def upload_insights_to_s3(
    destination_bucket: str,
    desired_file_extension: str = ".csv",
) -> None:
    """
    Example function for listing all insights in an account, and uploading any available
    to S3 as a CSV.
    """
    list_insights_query = Query()
    list_insights_query.get_insights()
    for insight in MC_CLIENT(list_insights_query).get_insights:
        report_name = str(Path(insight.name).with_suffix(desired_file_extension))

        if insight.available:
            report_url_query = Query()
            report_url_query.get_report_url(insight_name=insight.name, report_name=report_name)
            report_url = MC_CLIENT(report_url_query).get_report_url.url

            print(f"Uploading {report_name} to {destination_bucket}.")
            S3_CLIENT.upload_fileobj(
                Fileobj=requests.get(url=report_url, stream=True).raw,
                Bucket=destination_bucket,
                Key=report_name,
            )


if __name__ == "__main__":
    upload_insights_to_s3(destination_bucket="<BUCKET-NAME>")

See Monte Carlo's API reference for all supported queries and mutations.

For details and additional examples on how to map (convert) GraphQL queries to sgqlc operations please refer to the sgqlc docs.

Features

You can use pydoc to retrieve documentation on any feature packages (pydoc pycarlo.features).

For instance for circuit breakers:

pydoc pycarlo.features.circuit_breakers.service

Session configuration

By default, when creating a client the default profile from ~/.mcd/profiles.ini is used. This file created via montecarlo configure on the CLI. See Monte Carlo's CLI reference for more details.

You can override this usage by creating a custom Session. For instance, if you want to pass the ID and Token:

from pycarlo.core import Client, Session

client = Client(session=Session(mcd_id='foo', mcd_token='bar'))

Sessions support the following params:

• mcd_id: API Key ID.
• mcd_token: API secret.
• mcd_profile: Named profile containing credentials. This is created via the CLI (e.g. montecarlo configure --profile-name zeus).
• mcd_config_path: Path to file containing credentials. Defaults to ~/.mcd/.

You can also specify the API Key, secret or profile name using the following environment variables:

• MCD_DEFAULT_API_ID
• MCD_DEFAULT_API_TOKEN
• MCD_DEFAULT_PROFILE

When creating a session any explicitly passed mcd_id and mcd_token params take precedence, followed by environmental variables and then any config-file options.

Environment variables can be mixed with passed credentials, but not the config-file profile.

We do not recommend passing mcd_token as it is a secret and can be accidentally committed.

Authenticating with OAuth

Instead of an API key, you can authenticate with OAuth client credentials. Pass the OAuth client ID and secret along with your deployment's instance ID (e.g. us1, eu1):

from pycarlo.core import Client, Session

client = Client(session=Session(
    mcd_oauth_client_id='foo',
    mcd_oauth_client_secret='bar',
    mcd_instance_id='us1',
))

OAuth sessions support the following params:

• mcd_oauth_client_id: OAuth client ID.
• mcd_oauth_client_secret: OAuth client secret.
• mcd_instance_id: Deployment instance ID (e.g. us1, eu1), required for OAuth. Found under Account Information → Instance ID in the app.
• token_endpoint / oauth_api_endpoint: Optional overrides for the token and API endpoints (derived from the API endpoint automatically otherwise).

These can also be set via environment variables:

• MCD_DEFAULT_OAUTH_CLIENT_ID
• MCD_DEFAULT_OAUTH_CLIENT_SECRET
• MCD_DEFAULT_INSTANCE_ID

or stored in a profile via montecarlo configure --oauth. The SDK requests a short-lived access token and refreshes it automatically. As with API keys, prefer a profile or environment variables over passing the secret directly.

Integration Gateway API

There are features that require the Integration Gateway API instead of the regular GraphQL Application API, for example Airflow Callbacks invoked by the airflow-mcd library.

To use the Gateway you need to initialize the Session object passing a scope parameter and then use make_request to invoke Gateway endpoints:

from pycarlo.core import Client, Session

client = Client(session=Session(mcd_id='foo', mcd_token='bar', scope='AirflowCallbacks'))
response = client.make_request(
  path='/airflow/callbacks', method='POST', body={}, timeout_in_seconds=20
)

Advanced configuration

The following values also be set by the environment:

• MCD_VERBOSE_ERRORS: Enable logging. This includes a trace ID for each session and request.
• MCD_API_ENDPOINT: Customize the endpoint where queries and mutations are executed.

Enum Backward Compatibility

Unlike the baseline sgqlc behavior, this SDK is designed to maintain backward compatibility when new enum values are added to the Monte Carlo API. If the API returns an enum value that doesn't exist in your SDK version, it will be returned as a string with a warning logged, rather than raising an error. This allows older SDK versions to continue working when new features are added.

To avoid warnings and ensure full feature support, keep your SDK updated to the latest version.

References

• Monte Carlo App: https://getmontecarlo.com
• Product docs: https://docs.getmontecarlo.com
• Status page: https://status.getmontecarlo.com
• API (and SDK): https://apidocs.getmontecarlo.com
• CLI: https://clidocs.getmontecarlo.com

License

Apache 2.0 - See the Apache License 2.0 for more information.