Quickstart

Requirements

Python 3.10 - 3.13

We currently support the VSCode IDE and Jupyter Notebooks with our extension

Installation

In order to use the Etiq you need to install the Python package to your local environment.

Python Package

Install the etiq-copilot python package from PyPi:

pip install etiq-copilot

Usage

Use DebuggerCodeScanner to run a python file under Etiq's instrumentation. Pass the target source code to scan_code; Etiq executes the code, captures the runtime trace and observed objects, and returns a CodeScannerResult that can be used to inspect lineage outputs.

from pathlib import Path

from etiq_copilot.engine.implementations.scanner.code_scanner import DebuggerCodeScanner
from etiq_copilot.engine.implementations.scanner.scan_results import CodeScannerResult


def scan_file(scan_file_path: Path | str) -> CodeScannerResult:
    scan_file_path = Path(scan_file_path)
    original_code = scan_file_path.read_text(encoding="utf-8")
    scanner = DebuggerCodeScanner()
    return scanner.scan_code(code_str=original_code)

Example usage:

scan_results = scan_file("test_repo/iris_lineage_test.py")

Example Target Script

The quickstart uses this iris pipeline as the target script:

from sklearn import datasets
import sklearn.model_selection
from sklearn.ensemble import RandomForestClassifier
import pandas as pd

empty_dataframe = pd.DataFrame(columns=["a", "b"])

iris = datasets.load_iris()

iris_df = pd.DataFrame(iris.data, columns=iris.feature_names)

iris_df["target"] = iris.target

iris_train_df, iris_test_df = sklearn.model_selection.train_test_split(
    iris_df,
    test_size=0.2,
    random_state=31779,
)

amodel = RandomForestClassifier(random_state=0)

iris_training_features = iris_train_df[iris.feature_names].copy()
iris_test_features = iris_test_df[iris.feature_names].copy()
iris_target_training = iris_train_df["target"].copy()

amodel.fit(iris_training_features, iris_target_training)

iris_target_testing = iris_test_df["target"].copy()
preds = amodel.predict(iris_test_features)

Example Scan Output

After scanning the iris pipeline, inspect the scan result:

lineage_json = scan_results.create_full_lineage_graph(graph_format="json")

print("scan_errors:", scan_results.scan_errors)
print("dataframes:", scan_results.list_dataframes())
print("models:", scan_results.list_models())
print("agents:", scan_results.list_agents())
print("lineage_json:", lineage_json[:80] + "...")

Example output:

scan_errors: None
dataframes: ['iris_target_testing', 'iris_df', 'iris_training_features', 'preds', 'empty_dataframe', 'iris_target_training', 'iris_train_df', 'iris_test_features', 'iris_test_df']
models: ['amodel']
agents: []
lineage_json: {"objects": [{"style": "filled", "fillcolor": "#FFE18E", "shape": "circle"...

The full lineage_json value contains the generated graph. A shortened excerpt looks like this:

{
  "objects": [
    {
      "label": "preds",
      "shape": "circle",
      "fillcolor": "#FFE18E"
    },
    {
      "label": "amodel.predict",
      "shape": "diamond",
      "fillcolor": "#46A0FF"
    }
  ],
  "edges": [
    {
      "tail": 5,
      "head": 2
    },
    {
      "tail": 2,
      "head": 1
    }
  ]
}

The full lineage graph can be exported with create_full_lineage_graph(graph_format="json"). Generated node IDs can differ between runs.

Example Lineage Graph