DuckDB Connector

KumoRFM can connect directly to DuckDB databases, automatically inferring table metadata and relationships from the schema.

Installation

The DuckDB backend requires the ADBC DuckDB driver:

$ pip install kumoai[duckdb]

Quick Start

Connect to a file-based DuckDB database:

1 import kumoai.rfm as rfm
2 
3 graph = rfm.Graph.from_duckdb(connection="my_database.duckdb")

This will:

Connect to the DuckDB database
Discover all non-temporary, non-internal tables automatically
Infer column metadata (data types, semantic types, primary keys, time columns)
Detect foreign key relationships
Print a summary of the inferred metadata and links

Specifying Tables

Control which tables to include and customize their configuration:

1 graph = rfm.Graph.from_duckdb(
2     connection="my_database.duckdb",
3     tables=[
4         "USERS",
5         {"name": "ORDERS", "source_name": "ORDERS_SNAPSHOT"},
6         {"name": "ITEMS", "primary_key": "ITEM_ID"},
7     ],
8 )

Table configuration options:

Key	Description	Required
`name`	The table name used in PQL queries	Yes
`source_name`	The actual table name in the database (if different from `name`)	No
`primary_key`	Override the auto-detected primary key	No
`time_column`	The name of the time column for this table	No
`end_time_column`	The name of the end time column for this table	No
`columns`	Selected source columns or column specs, including expression columns	No

Connection Options

In-memory database:

1 from kumoai.rfm.backend.duckdb import connect
2 
3 conn = connect()  # in-memory DuckDB database
4 graph = rfm.Graph.from_duckdb(connection=conn)

From a file path:

1 graph = rfm.Graph.from_duckdb(connection="path/to/database.duckdb")

From an existing ADBC connection:

1 from kumoai.rfm.backend.duckdb import connect
2 
3 conn = connect("path/to/database.duckdb")
4 graph = rfm.Graph.from_duckdb(connection=conn)

From a connection config dict:

1 graph = rfm.Graph.from_duckdb(
2     connection={"uri": "path/to/database.duckdb"},
3 )

Controlling Metadata Inference

1 graph = rfm.Graph.from_duckdb(
2     connection="my_database.duckdb",
3     infer_metadata=False,
4     verbose=False,
5 )
6 
7 graph.infer_metadata()
8 graph.infer_links()

Manual Edge Specification

Override automatic link detection by providing edges explicitly:

1 graph = rfm.Graph.from_duckdb(
2     connection="my_database.duckdb",
3     edges=[
4         ("ORDERS", "USER_ID", "USERS"),
5         ("ORDERS", "ITEM_ID", "ITEMS"),
6     ],
7 )