API reference

DuckDB runtimeDownload PDF (2.8 MB)

This reference documents the helper library available inside the DuckDB runtime. Everything below is importable in jobs, datasets and notebooks without installing anything.

Conventions#

All connector methods are called on the session object returned by duckSession(). They share these conventions.

Convention Meaning
connName Name of a Groundzero connection. Credentials never appear in code.
returnFormat "duckdb" (default), "pandas" or "arrow".
targetDuckdbTable Optional name; the result is also registered under it so SQL can use it.
Lazy relations A "duckdb" result is a DuckDBPyRelation; it is evaluated when consumed.

What arrow returns

returnFormat="arrow" returns a pyarrow.RecordBatchReader from query() and from most connector reads, but a pyarrow.Table from readGzLakehouse. Call .read_all() on a reader to get a table, or use relation.to_arrow_table().

Session#

duckSession#

python
def duckSession(
    duckdbPath=":memory:", logLevel="INFO", threads=None, memory_limit=None,
    siteName=None, siteNames=None, **kwargs
) -> DuckSession: ...

Creates a configured session: opens a DuckDB connection, applies performance settings, attaches Groundzero Lakehouse catalogs for lakehouse connections and returns the session.

Parameter Type Default Description
duckdbPath str ":memory:" Database file path, or :memory:
logLevel str "INFO" Python log level name
threads int from compute Pins DuckDB threads for this session
memory_limit str from compute Pins DuckDB memory, e.g. "8GB"
siteName str or list None Lakehouse site for REST-backed connections
siteNames dict or list None Per-connection lakehouse sites
**kwargs any Extra DuckDB SET name='value' options

siteNames is preferably a mapping {"connectionName": "siteName"}. A list of {"connName": ..., "siteName": ...} entries is also accepted. A single site name applies to every REST-backed lakehouse connection; a list of site names must have one entry per connection or the mapping is ignored with a warning.

The session sets preserve_insertion_order=false, turns off the progress bar, turns on metadata caches for objects, HTTP and Parquet, and spills to /tmp/duckdb-spill.

Returns a DuckSession.

Raises AttributeError if logLevel is not a valid level name. Failures applying individual settings are logged as warnings, not raised.

python
from GZ.DuckDBUtils import duckSession

session = duckSession(logLevel="WARNING", threads=2, memory_limit="2GB")
print(session.query("SELECT current_setting('memory_limit') AS m",
                    returnFormat="pandas"))
session.close()

query#

python
def query(sqlQuery, params=None, returnFormat="duckdb"): ...

Runs SQL against everything registered in the session.

Parameter Type Default Description
sqlQuery str required SQL text
params list or dict None Values for prepared-statement parameters
returnFormat str "duckdb" "duckdb", "pandas" or "arrow"

Returns a relation, a pandas DataFrame or an Arrow reader. A statement that returns no result set (for example CREATE TABLE) returns the underlying DuckDB connection object.

Raises whatever DuckDB raises for invalid SQL (duckdb.ParserException, duckdb.CatalogException, ...); the error and the query text are logged first.

python
rows = session.query(
    "SELECT * FROM orders WHERE region = $r AND amount > $min",
    params={"r": "EMEA", "min": 100},
    returnFormat="pandas",
)

register and unregister#

python
def register(name, data): ...
def unregister(name): ...

register exposes a pandas DataFrame, Arrow table, relation or other object DuckDB can scan as a view called name. unregister removes it. Both return None and re-raise DuckDB errors.

listTables and describeTable#

python
def listTables() -> DataFrame: ...
def describeTable(tableName) -> DataFrame: ...

listTables returns a DataFrame with a name column. describeTable returns DuckDB's DESCRIBE output (column_name, column_type, null, key, default, extra). It raises ValueError("Invalid table name: ...") if the name contains characters other than letters, digits, underscores and dots.

registerLakehouseTable#

python
def registerLakehouseTable(connName, tableName, alias=None, materialize=True): ...

Reads a Groundzero Lakehouse table and keeps it in the session as lakehouse_<alias>.

Parameter Type Default Description
connName str required Lakehouse connection
tableName str required database.table name
alias str table name Suffix; dots and dashes become underscores
materialize bool True True copies into a table; False registers a view

Returns None. Raises the errors of readGzLakehouse.

close#

python
def close(): ...

Closes connector clients, unregisters session views and closes the DuckDB connection. Using the session as a with block calls it for you. Errors while closing are logged, not raised.

Groundzero Lakehouse#

readGzLakehouse#

python
def readGzLakehouse(
    connName, query=None, returnFormat="duckdb", tableName=None, siteName=None,
    filters=None
): ...

Aliases: executeGzLakehouse, readGZLAKEHOUSE.

Runs a read-only query on a lakehouse connection and returns the result.

Parameter Type Default Description
connName str required Groundzero Lakehouse connection
query str None Read-only SQL; tables as database.table
returnFormat str "duckdb" "duckdb", "pandas" or "arrow" (a Table)
tableName str None Reads SELECT * from this table when no query
siteName str None Site for this connection
filters list of dict None Structured predicates, see below

Each filter is {"column": str, "op": str, "values": list} with an optional "table". Operators: in, eq, between, gte, gt, lte, lt.

The call authorizes every referenced table for the caller, applies the caller's column masks and row policies by rewriting the query, runs it, and records an audit entry.

Returns a relation, DataFrame or Arrow table.

Raises

Exception When
ValueError Neither query nor tableName; non-read-only SQL; unparsable SQL; missing site name; unknown warehouse; site already bound
PermissionError Authentication fails, or the caller lacks Select on a table, or tables are not fully qualified
ConnectionError The lakehouse cannot be reached
RuntimeError Other lakehouse failures
python
df = session.readGzLakehouse(
    connName="lakehouse_main",
    query="SELECT region, count(*) AS n FROM retail.orders GROUP BY region",
    returnFormat="pandas",
)

Writing to the lakehouse#

DuckDB reads the lakehouse only. writeGzLakehouse (alias writeGZLAKEHOUSE) raises PermissionError so code written for another runtime fails clearly. Use the Spark runtime to write lakehouse tables, or write results to S3 or a database from DuckDB.

getLakehouseCatalog#

python
def getLakehouseCatalog(connName, siteName=None) -> str: ...

Attaches the connection's REST catalog if needed and returns its DuckDB catalog alias (of the form gzlakehouse__<connection>), so you can write fully qualified SQL against session.query:

python
catalog = session.getLakehouseCatalog("lakehouse_main", siteName="acme")
rel = session.query(f'SELECT count(*) FROM "{catalog}".retail.orders')

Raises ValueError if the connection has no lakehouse URL (it is not REST-backed) or no site can be determined.

Note

Queries issued through session.query against an attached catalog do not go through the authorization, masking and audit steps of readGzLakehouse. Use readGzLakehouse for governed access.

setLakehouseSite and configureGzLakehouseCatalogs#

python
def setLakehouseSite(connName, siteName): ...
def configureGzLakehouseCatalogs(connectionSiteNames=None, failFast=False): ...

setLakehouseSite binds a REST-backed connection to a site and attaches its catalog (ValueError if siteName is empty or the connection is not REST-backed). configureGzLakehouseCatalogs attaches the catalogs of every REST-backed connection; duckSession calls it for you. With failFast=False (default) an attach failure is deferred to first use.

Object storage#

readS3File#

python
def readS3File(
    connName, bucketName=None, filePath="", fileType="parquet", s3FullUrl=None,
    targetDuckdbTable=None, readOptions=None, returnFormat="duckdb"
): ...

Reads Parquet, CSV or JSON from Amazon S3 with DuckDB's native reader.

Parameter Type Default Description
connName str required S3 connection
bucketName str None Bucket; use with filePath
filePath str "" Key or glob inside the bucket
fileType str "parquet" parquet, csv or json (other values read as CSV)
s3FullUrl str None Full s3:// URL or glob; replaces bucket and path
targetDuckdbTable str None Register the result under this name
readOptions dict None Reader options, e.g. {"delim": ";"}
returnFormat str "duckdb" "duckdb", "pandas" or "arrow"

readOptions keys are passed to the DuckDB reader as named arguments, so they must be DuckDB's own option names (for CSV: delim, header, quote, dateformat, skip, and so on; delimiter is not valid). Booleans, numbers and strings are accepted. The call retries failed reads with all_varchar and then ignore_errors before raising the last error.

Raises ValueError for Either s3FullUrl or both bucketName and filePath must be provided or incomplete connection details, and DuckDB errors (for example duckdb.IOException) when the object cannot be read.

writeS3File#

python
def writeS3File(
    connName, bucketName=None, filePath="", fileType="parquet", s3FullUrl=None,
    sourceTable=None, sourceQuery=None, writeOptions=None
): ...

Alias: writeS3. Writes a table or query result with COPY ... TO.

Parameter Type Default Description
connName str required S3 connection
bucketName, filePath str None Destination, or use s3FullUrl
fileType str "parquet" parquet, csv or json
sourceTable str None Name of a table or view in the session
sourceQuery str None SQL whose result is written; wins over sourceTable
writeOptions dict None COPY options, e.g. {"compression": "zstd"}

Returns None. Raises ValueError for a missing destination (Either s3FullUrl or both bucketName and filePath must be provided), a missing source (Either sourceTable or sourceQuery must be provided) or an unsupported type (Unsupported file type: ...).

SQL databases#

executeJdbc#

python
def executeJdbc(
    connName, sqlQuery, params=None, targetDuckdbTable=None, returnFormat="duckdb"
): ...

Runs a query on a SQL database connection and loads the result.

Parameter Type Default Description
connName str required A PostgreSQL, MySQL, Oracle, SQL Server, Db2, Redshift, CockroachDB, SAP, Exasol, TimescaleDB or SQLite connection
sqlQuery str required Query in the database's own dialect
params list or dict None Bound parameters
targetDuckdbTable str None Register the result under this name
returnFormat str "duckdb" "duckdb", "pandas" or "arrow"

Returns the result in the requested format. Raises ValueError (Unsupported database type: ...) for connection types it cannot map, and the driver's errors for failing SQL. If the ADBC driver for PostgreSQL-compatible databases fails, the call falls back to SQLAlchemy and pandas automatically.

writeJdbc#

python
def writeJdbc(
    connName, targetTable, data=None, sourceTable=None, sourceQuery=None,
    mode="append"
): ...
Parameter Type Default Description
connName str required SQL database connection
targetTable str required Destination table
data relation, DataFrame or Table None Data to write
sourceTable str None Session table to write
sourceQuery str None Query to write
mode str "append" append or replace

Provide one of data, sourceTable or sourceQuery. PostgreSQL-compatible and MySQL targets are written through DuckDB's postgres and mysql extensions; other databases go through SQLAlchemy in chunks of 10,000 rows.

Returns True. Raises ValueError (Either data, sourceTable, or sourceQuery must be provided) or the database's errors. replace drops and recreates the table.

getSqlEngine#

python
def getSqlEngine(connName) -> sqlalchemy.Engine: ...

Returns a SQLAlchemy engine for a JDBC connection, for code that needs the native API. Dispose of it when you are done.

Cloud warehouses#

readSnowflake#

python
def readSnowflake(
    conn_name, sql_query, target_duckdb_table, warehouse=None,
    force_microsecond_precision=False, client_prefetch_threads=4,
    returnFormat="duckdb"
): ...

Streams a Snowflake query into a DuckDB table in Arrow batches, so Python holds only one batch at a time.

Parameter Type Default Description
conn_name str required Snowflake connection
sql_query str required Query to run
target_duckdb_table str required Table to create or replace
warehouse str None Overrides the connection's warehouse
force_microsecond_precision bool False Cast timestamps to microseconds; use for dates outside 1677-2262
client_prefetch_threads int 4 Download threads
returnFormat str "duckdb" "duckdb", "pandas", "arrow" or "row_count"

Raises ValueError for an unsupported returnFormat (Unsupported returnFormat. Use 'duckdb', 'pandas', 'arrow', or 'row_count'.) and Snowflake connector errors for failing SQL.

Outputs and run context#

gz_output and outputDF#

python
def gz_output(value): ...

Available as a global in datasets and jobs run by the runtime (it is not defined in notebooks). Records value as the run's output. If called repeatedly the last value is used. A variable named outputDF is the alternative; when both exist gz_output wins. See How-to guides, Return output and previews, for the accepted types and limits.

Placeholders#

${NAME} or {NAME} for STRTTM, ENDTM, CLNTNM, SITENM, PRJNM, JOBNM, RTE, RGN, USR and USRGRPS, and $NAME for job parameters passed through the external job API. See How-to guides, Use parameters and placeholders.

executeJob#

python
def executeJob(jobNames: list[str]) -> dict: ...

Import: from GZ.JobUtils import executeJob. Runs the named sub-jobs of the current multi-job run in sequence and returns {jobName: result}.

Parameter Type Description
jobNames list of str Names of sub-jobs to run

Raises ValueError if none of the names match the run's sub-jobs, and re-raises a sub-job's failure after recording it as failed.

sendEmail#

python
def sendEmail(subject, body, to_email, from_email, **kwargs): ...

Import: from GZ.SendEmail import sendEmail. Sends an email through the Groundzero notification service.

Parameter Type Description
subject str Subject line
body str Message body
to_email str Recipient address
from_email str Sender address
**kwargs any Extra fields added to the message

Returns None. Raises the error from the notification service.

python
from GZ.SendEmail import sendEmail

sendEmail(
    subject="Daily revenue ready",
    body="The 2026-09-30 revenue table was written.",
    to_email="data-team@example.com",
    from_email="noreply@example.com",
)

Clients and connection details#

getClient#

python
def getClient() -> ClientSession: ...

Import: from GZ.Client import getClient. Returns an object with one factory per connector that builds the vendor's native client from a connection, for the cases where the session methods are not enough.

Method Returns
getS3Client(connName) boto3 S3 client
getS3FSClient(connName) s3fs.S3FileSystem
getSQLEngine(connName) SQLAlchemy engine
getSnowflakeClient(connName) Snowflake connection
getMongoClient(connName) pymongo client
getBigQueryClient(connName) BigQuery client
getAzureBlobClient(connName) Azure Blob service client
getGCSClient(connName) Google Cloud Storage client
getDynamoDBClient(connName) DynamoDB client
getKinesisClient(connName) Kinesis client

The class also provides getSalesforceConnection, getCassandraSession, getSFTPClient, getRedshiftClient, getS3SelectClient, getCosmosDBClient, getElasticClient, getExasolConnection, getTimescaleClient, getPineconeClient, getTabularCatalog, getFireboltClient, getTeradataClient, getNeo4JClient, getInfluxDBClient, getDuckDBClient, getBusinessCentralClient, getShopifyClient, getMarketoClient, getAthenaClient, getCouchbaseClient, getClickhouseClient, getGoogleDriveClient, getOnedriveClient, getSplunkClient, getKDBAIClient, getDatabricksSQLClient, getAzureSynapseClient, getEventHubsClient, getSharepointClient, getScyllaDBClient, getAwsOpenSearchClient, getWeaviateClient, getFirestoreClient, getFirebaseDBClient, getOpenAIClient, getXAIClient, getAzureOpenAIClient, getAwsBedrockClient, getCohereClient, getTogetherAIClient, getFireworksAIClient, getHuggingFaceClient, getAnthropicClient, getGeminiClient and getMcpClient, each taking connName.

Raises ValueError when the named connection does not exist or has the wrong type, for example S3 connection details for 'x' not found or the connection type is incorrect.

Connectors#

The signatures below list every connector method on the session object. session object. Reads return the result in returnFormat; most connectors register the result as targetDuckdbTable when given. Writes return None, a status value or a count depending on the system, and raise the vendor's errors. Many connectors cache a client per connection; closeConnection(connName) releases it and session.close() releases all of them.

Object storage#

python
session.readS3File(
    connName, bucketName=None, filePath="", fileType="parquet", s3FullUrl=None,
    targetDuckdbTable=None, readOptions=None, returnFormat="duckdb"
)
session.writeS3File(
    connName, bucketName=None, filePath="", fileType="parquet", s3FullUrl=None,
    sourceTable=None, sourceQuery=None, writeOptions=None
)
session.writeS3(*args, **kwargs)

session.readGcsFile(
    connName, bucketName=None, filePath="", fileType="parquet", gcsFullUrl=None,
    targetDuckdbTable=None, readOptions=None, returnFormat="duckdb"
)
session.writeGcs(
    connName, bucketName=None, filePath="", fileType="parquet", gcsFullUrl=None,
    data=None, sourceTable=None, sourceQuery=None, writeOptions=None
)

session.readAzureBlob(
    connName, containerName, blobName, fileType="parquet", azureFullUrl=None,
    targetDuckdbTable=None, readOptions=None, returnFormat="duckdb"
)
session.writeAzureBlob(
    connName, containerName, blobName, fileType="parquet", azureFullUrl=None,
    data=None, sourceTable=None, sourceQuery=None, writeOptions=None
)

SQL databases and warehouses#

python
session.getSqlEngine(connName)
session.executeJdbc(
    connName, sqlQuery, params=None, targetDuckdbTable=None, returnFormat="duckdb"
)
session.writeJdbc(
    connName, targetTable, data=None, sourceTable=None, sourceQuery=None,
    mode="append"
)

session.getSnowflakeClient(conn_name, client_prefetch_threads=4)
session.readSnowflake(
    conn_name, sql_query, target_duckdb_table, warehouse=None,
    force_microsecond_precision=False, client_prefetch_threads=4,
    returnFormat="duckdb"
)

session.readBigQuery(
    connName, query, useLegacySql=False, targetDuckdbTable=None,
    returnFormat="duckdb", maxQueueSize=2, maxStreamCount=None
)
session.readBigQueryTable(
    connName, projectId, datasetId, tableId, selectedFields=None, rowFilter=None,
    targetDuckdbTable=None, returnFormat="duckdb", maxQueueSize=2,
    maxStreamCount=None
)
session.writeBigQuery(
    connName, tableName, data, writeDisposition="WRITE_APPEND", **kwargs
)

session.readAthena(
    connName, region, query, targetDuckdbTable=None, returnFormat="duckdb"
)
session.writeAthena(connName, region, sqlQuery)

session.executeDatabricks(
    connName, sourceQuery, targetDuckdbTable=None, returnFormat="duckdb",
    fetchBatchSize=10000
)
session.writeDatabricks(connName, sqlQuery)

session.readAzureSynapse(
    connName, query, targetDuckdbTable=None, returnFormat="duckdb",
    fetchBatchSize=10000
)

session.executeTeradata(
    connName, sourceQuery, targetDuckdbTable=None, returnFormat="duckdb",
    fetchBatchSize=10000
)
session.writeTeradata(connName, tableName, data, **kwargs)

session.getExasolClient(connName)
session.readExasol(
    connName, sqlQuery, targetDuckdbTable=None, returnFormat="duckdb",
    fetchBatchSize=10000
)
session.writeExasol(connName, tableName, data, **kwargs)

session.executeFirebolt(
    connName, sqlQuery, parameters=None, targetDuckdbTable=None,
    returnFormat="duckdb", fetchBatchSize=10000
)
session.getFireboltSystemInfo(
    connName, infoType="tables", targetDuckdbTable=None, returnFormat="duckdb"
)
session.writeFirebolt(connName, sqlQuery)

session.executeClickHouse(
    connName, sourceQuery, parameters=None, targetDuckdbTable=None,
    returnFormat="duckdb"
)
session.getSystemInfo(
    connName, table="tables", targetDuckdbTable=None, returnFormat="duckdb"
)
session.writeClickhouse(connName, tableName, data, database=None, **kwargs)

session.executeSingleStore(
    connName, sourceQuery, targetDuckdbTable=None, returnFormat="duckdb"
)
session.writeSingleStore(connName, tableName, data, **kwargs)

session.executeMotherDuck(
    connName, query=None, tableName=None, limit=None, alias="motherduck",
    targetDuckdbTable=None, returnFormat="duckdb"
)
session.readMotherDuck(
    connName, query=None, tableName=None, limit=None, alias="motherduck",
    targetDuckdbTable=None, returnFormat="duckdb"
)

NoSQL, search and graph#

python
session.getMongoClient(connName)
session.readMongo(
    connName, databaseName, collectionName, mongoQuery=None, projection=None,
    targetDuckdbTable=None, returnFormat="duckdb", **kwargs
)
session.writeMongodb(connName, databaseName, collectionName, data, **kwargs)

session.readDynamoDB(
    connName, tableName, filters=None, targetDuckdbTable=None,
    returnFormat="duckdb", parallelSegments=4
)
session.queryDynamoDB(
    connName, tableName, keyConditionExpression, expressionAttributeValues,
    expressionAttributeNames=None, filterExpression=None, indexName=None,
    targetDuckdbTable=None, returnFormat="duckdb"
)
session.writeDynamoDb(connName, tableName, data)

session.executeCosmos(
    connName, sourceQuery, enableCrossPartition=True, targetDuckdbTable=None,
    returnFormat="duckdb"
)
session.writeCosmosDb(connName, data, **kwargs)

session.executeCouchbase(
    connName, sourceQuery, parameters=None, targetDuckdbTable=None,
    returnFormat="duckdb"
)
session.writeCouchbase(connName, data, idField, **kwargs)

session.getCassandraSession(connName)
session.readCassandra(connName, cqlQuery, targetDuckdbTable=None, returnFormat="duckdb")

session.executeScyllaDB(
    connName, keyspace, sourceQuery, targetDuckdbTable=None, returnFormat="duckdb"
)

session.readFirestore(
    connName, collection, filters=None, orderBy=None, limit=None,
    targetDuckdbTable=None, returnFormat="duckdb"
)
session.writeFirestore(connName, collection, data, idField=None)

session.readRealtimeDatabase(
    connName, path, orderBy=None, limitToFirst=None, limitToLast=None,
    targetDuckdbTable=None, returnFormat="duckdb"
)

session.executeElasticsearch(
    connName, indexName, query=None, size=10000, targetDuckdbTable=None,
    returnFormat="duckdb"
)
session.executeESQL(connName, esqlQuery, targetDuckdbTable=None, returnFormat="duckdb")
session.writeElasticsearch(connName, indexName, data, idField=None, **kwargs)

session.searchOpenSearch(
    connName, index, query=None, size=1000, scroll=None, targetDuckdbTable=None,
    returnFormat="duckdb"
)
session.vectorSearch(
    connName, index, vector, vectorField, k=10, size=100, filter=None,
    targetDuckdbTable=None, returnFormat="duckdb"
)
session.getIndexMapping(connName, index, targetDuckdbTable=None, returnFormat="duckdb")
session.writeAwsOpenSearch(connName, index, data, idField=None, **kwargs)

session.executeSplunk(
    connName, searchQuery, earliestTime=None, latestTime=None, maxCount=0,
    targetDuckdbTable=None, returnFormat="duckdb"
)

session.streamGraphData(
    connName, graphName, nodeLabels=None, relationshipTypes=None,
    nodeProperties=None, relationshipProperties=None, targetDuckdbTable=None,
    returnFormat="duckdb"
)
session.executeCypherQuery(
    connName, cypherQuery, parameters=None, targetDuckdbTable=None,
    returnFormat="duckdb"
)
session.getGraphSchema(connName, targetDuckdbTable=None, returnFormat="duckdb")
session.getNodeLabels(connName, targetDuckdbTable=None, returnFormat="duckdb")
session.getRelationshipTypes(connName, targetDuckdbTable=None, returnFormat="duckdb")
session.writeNeo4j(connName, label, data, idField, properties=None, **kwargs)

session.executeInfluxQuery(
    connName, query, database=None, targetDuckdbTable=None, returnFormat="duckdb"
)
session.writeToInflux(
    connName, dataFrame, measurement, database=None, tagColumns=None,
    timeColumn="_time"
)

Streaming and lake formats#

python
session.readKafka(
    connName, topic, maxMessages=1000, autoOffsetReset="earliest",
    consumerTimeout=10000, targetDuckdbTable=None, returnFormat="duckdb"
)
session.writeKafka(connName, topic, data, keyField=None, **kwargs)

session.readEventHub(
    connName, consumerGroup="$Default", maxEvents=1000, startingPosition="-1",
    targetDuckdbTable=None, returnFormat="duckdb"
)
session.writeEventHub(connName, data, partitionKey=None, **kwargs)

session.readKinesis(
    connName, streamName=None, shardIteratorType="LATEST", limit=1000,
    maxShards=None, targetDuckdbTable=None, returnFormat="duckdb"
)
session.readKinesisStream(
    connName, streamName=None, shardIteratorType="LATEST", limit=1000,
    maxShards=None, targetDuckdbTable=None, returnFormat="duckdb"
)

session.readHudi(
    connName, tablePath, filters=None, columns=None, asOfTimestamp=None,
    targetDuckdbTable=None, returnFormat="duckdb"
)
session.readHudiIncremental(
    connName, tablePath, startCommitTime, endCommitTime=None, columns=None,
    targetDuckdbTable=None, returnFormat="duckdb"
)

SaaS, files and APIs#

python
session.executeREST(
    connName, endpoint=None, method=None, headers=None, params=None, body=None,
    resultPath=None, targetDuckdbTable=None, returnFormat="duckdb", timeout=60,
    maxPages=None, nextPath=None, pageParam=None, pageStart=1, maxRecords=None
)

session.getSalesforceConnection(connName)
session.readSalesforce(
    connName, soqlQuery, targetDuckdbTable=None, returnFormat="duckdb",
    useBulkApi=True, batchSize=10000, pkChunking=False, chunkSize=100000
)

session.readHubSpot(
    connName, objectType, properties=None, limit=100, after=None, filters=None,
    searchTerm=None, maxPages=None, rateLimit=10, targetDuckdbTable=None,
    returnFormat="duckdb"
)

session.readServiceNow(
    connName, table, query=None, limit=1000, displayValue=True, options=None,
    targetDuckdbTable=None, returnFormat="duckdb"
)
session.executeServiceNow(
    connName, sourceQuery, limit=1000, options=None, targetDuckdbTable=None,
    returnFormat="duckdb"
)

session.readZohoCRM(
    connName, module, perPage=200, maxPages=None, targetDuckdbTable=None,
    returnFormat="duckdb"
)
session.executeZohoCRM(connName, module, targetDuckdbTable=None, returnFormat="duckdb")

session.readShopify(
    connName, resource, limit=None, sinceId=None, fields=None,
    targetDuckdbTable=None, returnFormat="duckdb"
)
session.writeShopify(connName, resource, data, **kwargs)

session.readM365BC(
    connName, entity, filter=None, select=None, top=None, targetDuckdbTable=None,
    returnFormat="duckdb"
)
session.writeM365BC(connName, entity, data, **kwargs)

session.readGoogleDrive(
    connName, fileId, fileType=None, targetDuckdbTable=None, returnFormat="duckdb",
    **kwargs
)
session.listFiles(
    connName, folderId=None, query=None, targetDuckdbTable=None,
    returnFormat="duckdb"
)
session.writeGoogleDriveFile(
    connName, fileName, data, folderId=None, fileType="csv", **writeOptions
)

session.readOneDriveFile(
    connName, driveId, filePath, fileType="csv", targetDuckdbTable=None,
    returnFormat="duckdb", **readOptions
)
session.listOneDriveFiles(
    connName, driveId, folderPath=None, targetDuckdbTable=None,
    returnFormat="duckdb"
)
session.writeOneDriveFile(
    connName, driveId, filePath, data, fileType="csv", **writeOptions
)

session.readSharePointList(
    connName, siteUrl, listName, filter=None, select=None, top=None,
    targetDuckdbTable=None, returnFormat="duckdb"
)
session.writeSharePointList(connName, siteUrl, listName, data, idField=None)

session.readSFTPFile(
    connName, filePath, fileType="csv", targetDuckdbTable=None,
    returnFormat="duckdb", **readOptions
)
session.listSFTPFiles(
    connName, directory=".", targetDuckdbTable=None, returnFormat="duckdb"
)
session.writeSFTPFile(connName, filePath, data, fileType="csv", **writeOptions)

Vector stores#

python
session.queryVectors(
    connName, vector, topK=10, includeMetadata=True, includeValues=False,
    namespace=None, filter=None, targetDuckdbTable=None, returnFormat="duckdb"
)
session.upsertVectors(connName, vectors, namespace=None)
session.deleteVectors(connName, ids, namespace=None)
session.getIndexStats(connName, targetDuckdbTable=None, returnFormat="duckdb")
session.writePinecone(connName, data, namespace=None, **kwargs)

session.queryWeaviate(
    connName, className, properties=None, nearVector=None, nearText=None,
    where=None, limit=100, targetDuckdbTable=None, returnFormat="duckdb"
)
session.hybridSearch(
    connName, className, query, alpha=0.5, properties=None, where=None, limit=100,
    targetDuckdbTable=None, returnFormat="duckdb"
)
session.getSchema(connName, targetDuckdbTable=None, returnFormat="duckdb")
session.writeWeaviate(connName, className, data, batchSize=100, **kwargs)

session.readKDBAI(
    connName, tableName, filter=None, limit=None, targetDuckdbTable=None,
    returnFormat="duckdb"
)
session.searchKDBAI(
    connName, tableName, vectors, nResults=10, filter=None, targetDuckdbTable=None,
    returnFormat="duckdb"
)
session.writeKdbai(connName, tableName, data, **kwargs)

LLM providers#

Groundzero connections exist for Anthropic, Amazon Bedrock, Azure OpenAI, Cohere, Fireworks AI, Google Gemini, Hugging Face, OpenAI, Together AI and xAI. The generic methods below pick the provider from the connection's type, so code does not change when you switch providers.

complete#

python
def complete(
    connName, df, model=None, promptColumn="prompt", promptTemplate=None,
    systemPrompt=None, systemPromptColumn=None, outputColumn="llm_response",
    temperature=0.0, maxTokens=None, responseFormat=None, continueOnError=False,
    timeout=120, llmOptions=None, targetDuckdbTable=None, returnFormat="duckdb"
): ...

Adds an LLM completion to every row of a DataFrame or relation.

Parameter Type Default Description
connName str required LLM connection
df DataFrame or relation required Input rows
model str None Model name (optional)
promptColumn str "prompt" Column holding the prompt
promptTemplate str None Template using {column} fields instead of a prompt column
systemPrompt, systemPromptColumn str None Fixed or per-row system prompt
outputColumn str "llm_response" Name of the response column
temperature, maxTokens number 0.0, None Sampling controls
responseFormat str None Provider response format hint
continueOnError bool False Record row errors instead of raising
timeout int 120 Seconds per call
llmOptions dict None Provider-specific options

Besides outputColumn the result has <outputColumn>_status, _error, _provider, _model, _prompt_tokens, _completion_tokens, _total_tokens and _latency_ms columns.

Raises ValueError for an unknown connection (Connection 'x' was not found.), an unsupported connection type or missing input columns, and the provider's error for a failed call unless continueOnError=True.

python
import pandas as pd

tickets = pd.DataFrame({
    "ticket_id": [1, 2],
    "text": ["App crashes on login", "Love the new dashboard"],
})
scored = session.complete(
    connName="llm_openai",
    df=tickets,
    promptTemplate="Reply with one word, positive or negative: {text}",
    outputColumn="sentiment",
    continueOnError=True,
    returnFormat="pandas",
)

Fine-tuning and model management#

These generic methods route to the provider-specific method for the connection type (for example fineTuneOpenAI):

python
session.fineTune(connName, df, **kwargs)
session.distill(connName, df, **kwargs)
session.listFineTuneJobs(connName, **kwargs)
session.getFineTuneJob(connName, **kwargs)
session.cancelFineTuneJob(connName, **kwargs)
session.listModels(connName, **kwargs)

If the provider has no such operation the call raises NotImplementedError: <operation> is not available for connection type '<type>'.

Provider-specific methods exist for each operation. The suffix is the provider name: OpenAI, AzureOpenAI, AwsBedrock, Cohere, TogetherAI, HuggingFace, FireworksAI, Gemini, Anthropic or XAI.

Operation Methods (with provider suffix)
Fine-tune fineTune<P> (not XAI)
Distill distillOpenAI, distillAzureOpenAI, distillAwsBedrock
Jobs listFineTuneJobs<P>, getFineTuneJob<P>, cancelFineTuneJob<P> (Bedrock: stopFineTuneJobAwsBedrock)
Models listModels<P> (not Azure OpenAI)
Batch createBatchJob<P>, listBatchJobs<P>, getBatchJob<P>, cancelBatchJob<P>, getBatchResults<P>
Deployments createDeployment<P>, listDeployments<P>, getDeployment<P>, deleteDeployment<P>

Not every provider supports every operation: a method that does not apply to a provider fails with an explicit error (for example OpenAI has no deployment lifecycle). Fine-tuning methods take a pandas DataFrame with prompt and response columns and return a DataFrame describing the job:

python
job = session.fineTuneOpenAI(
    connName="llm_openai",
    df=training_df,
    model="gpt-4o-mini-2024-07-18",
    promptColumn="prompt",
    responseColumn="response",
)

MCP servers#

MCP connections let your code call tools, read resources and fetch prompts from a Model Context Protocol server. Every method returns the result as a relation (or the requested returnFormat).

python
session.listMcpTools(connName, targetDuckdbTable=None, returnFormat="duckdb")
session.callMcpTool(
    connName, toolName, arguments=None, targetDuckdbTable=None,
    returnFormat="duckdb"
)
session.listMcpResources(connName, targetDuckdbTable=None, returnFormat="duckdb")
session.readMcpResource(
    connName, uri, targetDuckdbTable=None, returnFormat="duckdb"
)
session.listMcpPrompts(connName, targetDuckdbTable=None, returnFormat="duckdb")
session.getMcpPrompt(
    connName, promptName, arguments=None, targetDuckdbTable=None,
    returnFormat="duckdb"
)
session.getMcpLlmConfig(
    connName, provider="openai", serverName=None, requireApproval="always",
    includeSecrets=False, targetDuckdbTable=None, returnFormat="duckdb"
)

completeWithMcpTool calls an MCP tool for each row and then asks an LLM to complete a prompt that includes the tool result:

python
session.completeWithMcpTool(
    llmProvider, llmConnName, mcpConnName, df, toolName, arguments=None,
    argumentsColumn=None, promptColumn="prompt", promptTemplate=None,
    mcpPromptTemplate=None, systemPrompt=None, systemPromptColumn=None,
    outputColumn="llm_response", mcpResultColumn="mcp_tool_result",
    mcpErrorColumn="mcp_tool_error", model=None, temperature=0.0,
    maxTokens=None, responseFormat=None, continueOnError=False, timeout=120,
    llmOptions=None, targetDuckdbTable=None, returnFormat="duckdb"
)

MCP connection setup

Create the connection of type MCP Server with transport streamable_http, a server URL that includes http:// or https://, an authentication type (bearer with a token, or none for public servers), optional headers and allowed tools, and a timeout. A URL without a scheme fails before tool discovery.

PDF extraction#

python
session.parsePDF(
    s3ConnName=None, inputUri=None, outputPrefix=None, outputFormat="csv",
    outputFormats=None, inputFile=None, connName=None, hybrid=False,
    ocrMode="auto", ocrStack="ppocrv6", strict=False, password=None,
    language="eng", maxPages=None, maxFileBytes=None, llmProvider=None,
    llmConnName=None, llmModel=None, llmBaseUrl=None, llmApiKey=None,
    outputS3ConnName=None, returnAgentPages=False, contentAddressed=False,
    inlineAgentPages=True, pageRange=None
)

Alias: extractPDF. Parses one document from S3 and writes the results (CSV, JSON, XLSX or Markdown) under outputPrefix.

Parameter Type Default Description
s3ConnName str required* S3 connection holding the input (connName is an alias)
inputUri str required* Document key or s3:// URI (inputFile is an alias)
outputPrefix str generated Destination prefix
outputFormat, outputFormats str, list "csv" One or several output formats
hybrid bool False Hybrid extraction; pairs with the llm... options
ocrMode str "auto" OCR policy
ocrStack str "ppocrv6" OCR stack
strict bool False Strict extraction mode
password str None Password for encrypted PDFs
maxPages, maxFileBytes int None Safety limits
returnAgentPages bool False Produce page JSON or Markdown for Agents
pageRange tuple None (first, last) page; requires returnAgentPages=True

*One of inputUri or inputFile is required, and an S3 connection name.

Returns a dictionary that includes result_uri. Raises ValueError (s3ConnName is required, inputUri or inputFile is required, pageRange requires returnAgentPages=True) and PdfLimitError when a source is larger than the configured limit.

python
session.readPDF(
    s3ConnName, inputUri=None, outputPrefix=None, outputFormat="csv",
    outputFormats=None, targetDuckdbTable=None, readOptions=None,
    returnFormat="duckdb", outputS3ConnName=None, **options
)
session.readPDFResult(
    resultUri, outputFormat="csv", s3ConnName=None, targetDuckdbTable=None,
    readOptions=None, returnFormat="duckdb"
)
session.parsePDFCorpus(
    s3ConnName, sourcePrefix=None, inputUris=None, outputPrefix=None,
    outputS3ConnName=None, pageRangeSize=64, partitions=None, ocrMode="auto",
    language="eng", strict=False, password=None, maxFileBytes=None,
    maxDocuments=10000
)

readPDF parses a document and returns its table in one call (pdfParser, readPDFExtraction are aliases); readPDFResult (alias readPDFExtractionResult) loads an already parsed csv or json result and raises ValueError for other formats. parsePDFCorpus (alias processPDFCorpus) parses every PDF under sourcePrefix, or the listed inputUris, with restartable, content addressed output. partitions sets the number of I/O workers (1 to 8); OCR still processes one document model at a time.

Groundzero Agents#

python
def listAgents(connName) -> list[dict]: ...

def invokeAgent(
    connName, agentId, message, sessionId=None, *, requestId=None,
    expectedAgentVersion=None, disableTools=False, pdfSourceConnName=None,
    pdfSourceObject=None, pdfSinkConnName=None, pdfSinkPrefix=None
) -> dict: ...

def invokeAgentPDFCorpus(
    connName, agentId, message, pdfSourceConnName, pdfSinkPrefix,
    pdfSourcePrefix=None, pdfSourceObjects=None, pdfSinkConnName=None,
    pageRangeSize=64, partitions=None, summaryWorkers=4,
    summaryInputChars=100000, maxDocuments=10000, maxAgentCalls=50000,
    ocrMode="auto", language="eng", strict=False, password=None,
    maxFileBytes=None
) -> dict: ...
Method Purpose
listAgents Agents visible to the connection's service account
invokeAgent Run an Agent in the background and wait for the final result (polls up to 15 minutes)
invokeAgentPDFCorpus Extract JSON or cited facts from every complete PDF

Parameters worth noting:

Parameter Description
agentId Agent identifier (a UUID)
message Instruction for the Agent; must be a non-empty string
expectedAgentVersion Positive integer; the call fails if the approved version differs
disableTools True runs without tools; it cannot be combined with PDF arguments
pdfSourcePrefix / pdfSourceObjects Supply exactly one
summaryWorkers Concurrent document workflows, 1 to 32
maxDocuments, maxAgentCalls Bounds for one run, defaults 10,000 and 50,000

Raises ValueError for invalid arguments (agentId is required, message is required, expectedAgentVersion must be a positive integer), RuntimeError when an Agent run returns an invalid run identifier or status, and TimeoutError when polling exceeds 15 minutes. A timeout does not cancel the run: inspect it before starting another.

Data quality and record linkage#

python
session.createDataQualityChecker(data, connectionName=None, config=None,
                                 references=None)
session.runDataQualityChecks(data, connectionName=None, config=None,
                             includeLlm=True, references=None)
session.runTableDataQualityChecks(tableName, connectionName=None, config=None,
                                  includeLlm=True, references=None)
session.runQueryDataQualityChecks(query, connectionName=None, config=None,
                                  includeLlm=True, references=None)
session.saveDataQualityReport(report, path, reportFormat="json")
Parameter Description
data DuckDB relation, Arrow table or reader, pandas or Polars DataFrame
tableName, query A session table, or SQL, to check without copying
connectionName Groundzero LLM connection used for the narrative
config A qualipilot.QualipilotConfig
includeLlm False runs only deterministic checks
references Optional reference data for referential checks
reportFormat json, html or markdown

The run methods return a Qualipilot QualityReport. Checks reuse the session's DuckDB connection, and only a compact aggregate of the deterministic results is sent to the LLM, not row samples.

Local record linkage and consolidation collect inputs into Polars frames, so each requires maxRows:

python
def runLocalRecordLinkage(data, config, rightData=None, *, maxRows): ...
def runLocalRecordDeduplication(data, config, consolidation, *, maxRows): ...
def consolidateLocalRecords(data, idColumn, clusters, config, *, maxRows): ...

config is a qualipilot.linking.LinkConfig (mode dedupe or link) and consolidation a ConsolidationConfig. Raises TypeError if maxRows is not an integer, ValueError if it is not positive, if an input has more than maxRows rows (input has more than maxRows=N rows), if link mode has no rightData, if dedupe mode has rightData, or if deduplication is not given a dedupe configuration. maxRows bounds rows, not width or candidate pairs: select only the columns you need and keep blocking rules tight.

Talend migration helpers#

Import from GZ.TalendCompilerUtils. These helpers load an exported Talend project into a DuckDB manifest table, parse it and compile it into Groundzero job code. They take the session as their first argument.

Function Purpose
loadTalendProjectFromS3ToDuckDB(duck, connName, prefix, ...) Load project files from S3 into a manifest table
loadTalendProjectFilesToDuckDB(duck, files, ...) Load files you already hold in memory
parseTalendManifestToIR(duck, ...) Parse the manifest into an intermediate representation
compileTalendManifestToGroundZeroJobs(duck, ...) Parse and compile to job code
compileTalendIRToGroundZeroJobs(ir, ...) Compile an existing representation
writeTalendCompileResultToDuckDB(duck, result, ...) Persist compiled jobs to a table
renderTalendCompileReport(result) Render a compact report