Gwenlake Python Library

The Gwenlake Python library provides convenient access to the Gwenlake API from applications written in Python. A single Gwenlake client gives you access to your catalog — projects, datasets, files and SQL.

Installation

pip install -U gwenlake

Or install the latest development version straight from GitHub:

pip install -U git+https://github.com/gwenlake/gwenlake-python

Authentication

The client authenticates with a Bearer token, resolved in this order:

1. an explicit api_key / credentials passed to the client,
2. a named profile,
3. the GWENLAKE_API_KEY environment variable,
4. the default profile in ~/.gwenlake/credentials.

export GWENLAKE_API_KEY='sk-...'

from gwenlake import Gwenlake

# uses GWENLAKE_API_KEY, or the default ~/.gwenlake/credentials profile
client = Gwenlake()

# or pass the key explicitly
client = Gwenlake(api_key="sk-...")

# or pick a profile from ~/.gwenlake/credentials
client = Gwenlake(profile="myteam")

The ~/.gwenlake/credentials file is an INI file with one section per profile, holding either a static token (API key) or OAuth2 client_id / client_secret.

Projects

projects = client.projects.list()
for p in projects:
    print(p["alias"], p["id"])

project = client.projects.get("res.project.…")

Datasets

datasets = client.datasets.list()
for d in datasets:
    print(d["alias"], d["id"])

dataset = client.datasets.get("res.dataset.…")

Files

Files live inside a dataset.

dataset_id = "res.dataset.…"

# list files
for f in client.files.list(dataset_id):
    print(f["filename"], f["file_size"])

# upload a local file (optionally into a subdirectory with path=...)
client.files.upload(dataset_id, "report.pdf")
client.files.upload(dataset_id, "report.pdf", path="docs")

# download a file
content = client.files.download(dataset_id, "report.pdf")

# presigned URL / delete
url = client.files.presigned_url(dataset_id, "report.pdf")
client.files.delete(dataset_id, "report.pdf")

SQL

Run SQL against a dataset (DuckDB), referencing it as '<project_alias>.<dataset_alias>'. With format="json" the rows are returned under data:

result = client.statements.create(
    statement="SELECT * FROM 'flights.flight-data' LIMIT 10",
    format="json",
)
for row in result["data"]:
    print(row)

Pass a connection_id to run the statement against a connection's native engine (PostgreSQL, S3, …) instead of a dataset.

Transforms

A transforms layer (gwenlake.transforms) lets you write dataset-to-dataset transformations as decorated functions. Datasets (and models) are addressed as "<project_alias>.<alias>" — the same handle used in SQL.

transform_df — the function receives each Input as a pandas.DataFrame and returns the DataFrame to write to the (single) Output. The result is written automatically (snapshot/replace by default):

from gwenlake.transforms import transform_df, Input, Output

@transform_df(
    raw_data=Input("Project_A.users"),
    processed_data=Output("Project_A.users_filtered"),
)
def process(raw_data):
    df = raw_data[raw_data["age"] >= 18].copy()
    df["name_upper"] = df["name"].str.upper()
    return df

process(client)   # reads, computes, writes

With several Outputs, return a dict keyed by their keyword names; each DataFrame is written to the matching dataset:

@transform_df(
    raw=Input("Project_A.contacts"),
    valid=Output("Project_A.contacts_valid"),
    rejected=Output("Project_A.contacts_rejected"),
)
def split(raw):
    ok = raw["email"].str.contains("@")
    return {"valid": raw[ok], "rejected": raw[~ok]}

transform — the lower-level form: the function receives TransformInput / TransformOutput objects and reads/writes explicitly. Use it for non-tabular data (images, PDFs, …) via .filesystem():

from gwenlake.transforms import transform, Input, Output

@transform(
    my_input=Input("Project_A.users"),
    my_output=Output("Project_A.users_distinct"),
)
def dedupe_users(my_input, my_output):
    df = my_input.dataframe()
    # mode="replace" (default) clears the dataset first; "append" keeps existing files
    my_output.write_dataframe(df.drop_duplicates(), mode="replace")

@transform(
    images=Input("Project_A.scans"),
    thumbnails=Output("Project_A.scans_processed"),
)
def process_files(images, thumbnails):
    src, dst = images.filesystem(), thumbnails.filesystem()
    for entry in src.ls():
        data = src.read(entry["filename"])          # raw bytes (PDF, image, …)
        with dst.open(f"copy/{entry['filename']}", "wb") as f:
            f.write(data)

Connections

A connection is a datasource a project declares once (an SMB share, a bucket, a SharePoint library). A file sync copies what it holds into a dataset; a transform that declares the connection as a Connection(...) input reads it itself — walks the directory, picks the files it wants, lays its output out as it likes. .filesystem() is read-only and goes through the catalog as the build's own identity, so the connection's credentials never reach the code:

import shutil
from gwenlake.transforms import transform, Connection, Output

@transform(
    share=Connection("Ops.drop-folder"),          # "<project>.<connection slug>"
    out=Output("Ops.collected"),
)
def collect(share, out):
    fs_in, fs_out = share.filesystem(), out.filesystem()
    for f in fs_in.ls(glob="**/*.pdf"):           # f["path"] = "2024/q3/report.pdf"
        with fs_in.open(f["path"], "rb") as src, fs_out.open(f["path"], "wb") as dst:
            shutil.copyfileobj(src, dst)

ls(path=, glob=) lists recursively (paths relative to the connection's root; * stays within a directory, ** descends — the same rule as a sync's path_glob), read(path) / open(path, "rb") fetch one file. The dataset FileSystem.ls takes the same glob= so an output can be walked the same way. The catalog scanner reads Connection(...) like Input(...), so the lineage shows the connection feeding the output. client.connections exposes the same calls directly (list, get, list_files, iter_files, download).

Incremental transforms

A transform that must not redo what it has already done keeps its memory in one of its own Outputs — a log of the files it has processed — and reads it back on the next run. An Output can never be an Input of the same transform (that is a cycle in the lineage), so @incremental — stacked above @transform / @transform_df — gives the Output a "previous" reading mode and makes writes add to it:

import pandas as pd
from gwenlake.transforms import incremental, transform, Connection, Output

@incremental(semantic_version=1)
@transform(
    share=Connection("Ops.drop-folder"),
    pages=Output("Ops.pages"),
    log=Output("Ops.processed"),
)
def ingest(share, pages, log):
    done = set(log.dataframe("previous", schema=["path"])["path"])   # empty on the first run
    new = [f for f in share.filesystem().ls(glob="**/*.pdf") if f["path"] not in done]
    if not new:
        return
    pages.write_dataframe(parse(new))                                  # appended
    log.write_dataframe(pd.DataFrame({"path": [f["path"] for f in new]}))   # last: a failure re-processes, never forgets

dataframe("previous", schema) is the output as the last successful run left it (read it before the run's first write; an empty frame of schema while there is none), dataframe("current") re-reads it, filesystem("previous") is the same for files. Writes default to modify (set_mode("replace") or mode="replace" snapshots instead). Bump semantic_version to start over on the next build. An Iceberg input also answers dataframe("added") inside a build — the rows appended since the last successful run, read from the table's own snapshot history up to the point pinned before the run — and dataframe("previous"), the table as the last run saw it. A files dataset or a connection keeps no history: those are always read whole, and "added" is refused rather than quietly answered with everything — the log Output is what remembers. A table rewritten in between (snapshot, delete, upsert) has no delta either; the catalog says so, and the run reads it whole or bumps semantic_version.

Models

A model is a catalog resource whose artifacts live in the git repository the code lives in (/models in the catalog). train produces one, and Model(...) binds one a transform loads — conventionally as model=:

import joblib
from gwenlake.transforms import train, transform_df, Input, Model, Output

@train(
    training_set=Input("Project_A.churn_training"),
    output=Output("Project_A.churn"),          # an Output of @train is a MODEL
)
def fit(training_set, output):
    clf = fit_classifier(training_set)         # training_set is a DataFrame
    joblib.dump(clf, output.file("model.pkl")) # write under the model's directory
    return {"auc": 0.91}                       # returned dict -> the model's metrics

@transform_df(
    customers=Input("Project_A.customers"),
    model=Model("Project_A.churn"),            # a model this transform loads
    output=Output("Project_A.churn_scores"),
)
def predict(customers, model):
    clf = joblib.load(model.file("model.pkl"))
    return customers.assign(churn=clf.predict(customers))

Models are a catalog resource served by api-catalog. Its /models endpoints are not routed through the public gateway today (that path serves the inference model list), so model.info() / model.update() work from inside a build — where the client already points at api-catalog — but not against api.gwenlake.com. model.path never needs an API call during a build.

output.path (and model.path) is the model's directory in the checkout: during a build the engine sets it, commits whatever the training run wrote there, and pins that commit as the model's version. model.parameters, model.version and model.update(metrics=..., version=...) cover the model card. Lineage follows: datasets -> train -> model -> transform -> dataset.

Large datasets — page through with LIMIT/OFFSET instead of loading everything at once. iter_dataframes() yields pandas.DataFrame chunks and write_dataframes() streams them back out as part-00000.parquet, …:

@transform(
    big_dataset=Input("Project_A.events"),
    result=Output("Project_A.events_clean"),
)
def transform_in_chunks(big_dataset, result):
    chunks = (
        chunk[chunk["valid"]]
        for chunk in big_dataset.iter_dataframes(chunk_size=50_000, order_by="id")
    )
    result.write_dataframes(chunks, mode="replace")

Pass order_by= for a deterministic page split. The transforms layer is synchronous.

Async

Every resource is also available on AsyncGwenlake:

import asyncio
from gwenlake import AsyncGwenlake

async def main():
    client = AsyncGwenlake()
    print(await client.projects.list())

asyncio.run(main())

See examples/ for runnable scripts.