Create a Credential and Source

Overview

Creating a resource requires mutations, because it modifies data on the server rather than only reading it. Create a credential, and use that credential to infer the schema for a given dataset. Then, use the credential and schema to create a source.

Steps

Create a credential

Create a credential and pass the input as a dictionary. The type-specific fields (the GCP service account credential and the catalog flag) go under config.gcp. Cross-reference the API documentation to see which variables CredentialCreateInput expects and requires.

Infer the schema

Instead of writing out the schema for the source by hand, have Validio infer it. This is what happens under the hood when you set up a source in the UI.

Create the source

Create a source with the credential from the first step and the schema inferred in the second step. The type-specific fields (project, dataset, table, and schedule) go under config.gcpBigQuery. Pass all the values for SourceCreateInput to create the source.

Full code sample

Python

import os
from typing import Any

from validio_sdk.client import Client


async def example():
    client = Client()
    credential_id = await create_gcp_credential(client)
    source_id = await create_bigquery_source(client, credential_id)

    print(f"Created source with id {source_id}")


async def create_gcp_credential(client: Client) -> str:
    query = """
    mutation CredentialCreate($input: CredentialCreateInput!) {
      credentialCreate(input: $input) {
        id
      }
    }
    """

    variable_values = {
        "input": {
            "name": "my-credential",
            "namespaceId": os.environ["EXAMPLE_NAMESPACE_ID"],
            "config": {
                "gcp": {
                    "credential": os.environ["EXAMPLE_GCP_CREDENTIAL"],
                    "enableCatalog": True,
                },
            },
        }
    }
    result = await client.execute(query, variable_values=variable_values)

    return result["credentialCreate"]["id"]


async def get_jtd_schema(client: Client, credential_id: str) -> dict[str, Any]:
    query = """
    query GcpBigQueryInferSchema($input: GcpBigQueryInferSchemaInput!) {
      gcpBigQueryInferSchema(input: $input)
    }
    """
    variable_values = {
        "input": {
            "credentialId": credential_id,
            "project": "some-project",
            "dataset": "some-dataset",
            "table": "some-table",
        }
    }

    result = await client.execute(
        query,
        variable_values=variable_values,
    )

    return result["gcpBigQueryInferSchema"]


async def create_bigquery_source(client: Client, credential_id: str) -> str:
    jtd_schema = await get_jtd_schema(client, credential_id)

    create_source_query = """
    mutation SourceCreate($input: SourceCreateInput!) {
      sourceCreate(input: $input) {
        id
      }
    }
    """
    variable_values = {
        "input": {
            "name": "my-bigquery-source",
            "credentialId": credential_id,
            "jtdSchema": jtd_schema,
            "config": {
                "gcpBigQuery": {
                    "project": "some-project",
                    "dataset": "some-dataset",
                    "table": "some-table",
                    "schedule": "0 0 * * *",
                },
            },
        }
    }
    result = await client.execute(create_source_query, variable_values=variable_values)

    return result["sourceCreate"]["id"]

Overview

Getting Started

Validio API

Validio CLI

Validio SDK

Validio Code (IaC)

Overview

Steps

Full code sample

​Overview

​Steps

​Full code sample

Overview

Steps

Full code sample