relationalai.Provider

class Provider(
    profile: str|None = None,
    config: Config|None = None,
    connection: Session|None = None
)

The Provider class is used to manage the RAI Native App from Python. It provides methods corresponding to SQL procedures and RAI CLI commands.

Parameters

Name	Type	Description
profile	str	Optional name of the raiconfig.toml profile to use when creating the model. If None, the active raiconfig.toml profile is used.
config	Config	Optional configuration object to use when creating the model. If None, configuration details provided by the profile and connection parameters are used.
connection	Session	Optional Snowpark Session object to use when creating the model. If None, connection details provided by the profile or config parameters are used.

Example

By default, Provider uses the active profile in your raiconfig.toml file:

import relationalai as rai

# Create a Provider object using the active raiconfig.toml profile.
app = rai.Provider()

# Create a Provider object using a specific raiconfig.toml profile.
app = rai.Provider(profile="<profile_name>")

If you are not using a raiconfig.toml file, you can provide configuration details directly to the Provider object by passing a Config object or a Snowpark Session object to the config or connection parameter, respectively:

# Create a Provider object using a Config object instead of a raiconfig.toml file.
config = rai.Config({
    "platform": "snowflake",
    "authenticator": "snowflake",
    "account": "<SNOWFLAKE_ACCOUNT>",
    "user": "<SNOWFLAKE_USER>",
    "password": "<SNOWFLAKE_PASSWORD>",
    "role": "<SNOWFLAKE_ROLE>",
    "warehouse": "<SNOWFLAKE_WAREHOUSE>",
})
app = rai.Provider(config=config)


# Create a Provider object using a Snowpark Session object.
from snowflake.snowpark import Session

connection_params = {
    "account": "<SNOWFLAKE_ACCOUNT>",
    "user": "<SNOWFLAKE_USER>",
    "password": "<SNOWFLAKE_PASSWORD>",
    "role": "<SNOWFLAKE_ROLE>",
    "warehouse": "<SNOWFLAKE_WAREHOUSE>",
}
session = Session.builder.configs(connection_params).create()
app = rai.Provider(connection=session)

Using a Session object is useful when you need to authenticate with Snowflake using a method other than a username and password, such as SSH. See the Snowpark docs for more information on creating Session objects.

Methods

Name	Description	Returns
.activate()	Activate the RAI Native App.	None
.create_streams()	Create streams in a model.	None
.deactivate()	Deactivate the RAI Native App.	None
.delete_stream()	Delete a stream in a model.	None
.list_streams()	List all streams in a model.	list[dict]
.sql()	Execute a SQL query.	Snowpark DataFrame

.activate()

Provider.activate() -> None

Activate the RAI Native App and block until the activation is complete. Requires the Snowflake user set in your configuration to have the app_admin application role.

Returns

None

Example

Use .activate() to activate the RAI Native App if it is not already active:

import relationalai as rai

app = rai.Provider()
app.activate()

Note

Activating the RAI Native App provisions several Snowflake and RAI resources and may take several minutes to complete. See RAI Native App Status for details.

.create_streams()

Provider.create_streams(sources: list[str], model:str, force:bool=False) -> None

Creates data streams from one or more Snowflake tables or views to a RelationalAI (RAI) model. At least one engine with a READY state must be available to create a data stream. You must have SELECT privileges and change tracking must be enabled on the source object. See Supported Column Types for a list of column types supported in a data stream’s source table or view. Requires the cdc_admin application role.

Stream data from Snowflake tables into a model and block until the streams are ready. Requires the Snowflake user set in your configuration to have the cdc_admin application role.

Parameters

Name	Type	Description
sources	list[str]	List of Snowflake tables to stream data from. Table names must be fully qualified with the Snowflake database and schema names, e.g. "<db_name>.<schema_name>.<table_name>".
model	str	Name of the model to stream data into.
force	bool	Whether to force the recreation of streams if they already exist. (Default: False)

Returns

None

Example

Use .create_streams() to stream data from Snowflake tables into a model:

import relationalai as rai

# Create a model.
model = rai.model("MyModel")

# Stream data from Snowflake tables into the model.
app = rai.Provider()
app.create_streams(
    model=model.name,
    sources=[
        "my_db.my_schema.my_table1",
        "my_db.my_schema.my_table2",
    ]
)

# After streams are created, you can use them as sources for types.
Type1 = model.Type("Type1", source="my_db.my_schema.my_table1")
Type2 = model.Type("Type2", source="my_db.my_schema.my_table2")

See Data Management for more information on managing data streams.

.deactivate()

Provider.deactivate() -> None

Deactivate the RAI Native App. Requires the Snowflake user set in your configuration to have the app_admin application role.

Returns

None

Example

Use .deactivate() to deactivate the RAI Native App while it is not in use:

import relationalai as rai

app = rai.Provider()
app.activate()

Caution

While the RAI Native App is deactivated, RAI models cannot be created or queried. See RAI Native App Status for details.

.delete_stream()

Provider.delete_stream(stream_id: str, model:str) -> None

Delete a stream from a model. Requires the Snowflake user set in your configuration to have the cdc_admin application role.

Parameters

Name	Type	Description
stream_id	str	The name of the stream to delete, e.g. "<db_name>.<schema_name>.<table_name>".
model	str	Name of the model to delete the stream from.

Returns

None

Example

Use .delete_stream() to delete a stream from a model:

import relationalai as rai

app = rai.Provider()
app.delete_stream("my_db.my_schema.my_table", model="MyModel")

See Data Management for more information on managing data streams.

.list_streams()

Provider.list_streams(model:str) -> None

List all streams in a model. Requires the Snowflake user set in your configuration to have the cdc_admin application role.

Parameters

Name	Type	Description
model	str	Name of the model to list streams from.

Returns

A list of dict objects representing the streams in the model.

Example

Use .list_streams() to list all streams in a model:

from pprint import pprint

import relationalai as rai

app = rai.Provider()
pprint(app.list_streams("MyModel"))
# [{'batches': '1',
#   'created': datetime.datetime(2024, 9, 23, 4, 56, 20, 229000),
#   'creator': 'user@company.com',
#   'errors': None,
#   'id': 'ds_00d1d5cc_71d0_44f0_9382_b2b72349cb0b',
#   'model': 'MyModel',
#   'name': 'MY_DB.MY_SCHEMA.MY_TABLE1',
#   'status': 'LOADED'},
#  {'batches': '1',
#   'created': datetime.datetime(2024, 9, 23, 4, 56, 3, 325000),
#   'creator': 'user@company.com',
#   'errors': None,
#   'id': 'ds_34ea56e7_d1d9_4cb0_beba_4a3a6ece060e',
#   'model': 'MyModel',
#   'name': 'MY_DB.MY_SCHEMA.MY_TABLE2',
#   'status': 'LOADED'}]

.sql()

Provider.sql(
    query: str,
    params: list[Any]|None = None,
    format: str = "list",
)

Execute a SQL query and return the results as a list of Snowpark Row objects. Set format to "lazy" to return a SnowPark DataFrame, or "pandas" to return a pandas DataFrame.

Parameters

Name	Type	Description
query	str	The SQL query to execute.
params	list[Any]	Optional list of binding parameters to pass to the query.
format	str	The format of the returned DataFrame. Must be one of: "list" (Default) Return a list of Snowpark Row objects. "lazy" Return a Snowpark DataFrame. "pandas" Return a pandas DataFrame.

Returns

A list of Snowpark Row objects, a Snowpark DataFrame, or a pandas DataFrame depending on the format parameter.

Example

Use .sql() to execute a SQL query and return the results as a Snowpark DataFrame:

import relationalai as rai

app = rai.Provider()
app.sql(
    query="SELECT * FROM values (?, ?), (?, ?)",
    params=[1, "a", 2, "b"],
)
# [Row(COLUMN1=1, COLUMN2='a'), Row(COLUMN1=2, COLUMN2='b')]

Change the format parameter to change the return type:

# Return a Snowpark DataFrame.
df = app.sql(
    query="SELECT * FROM values (?, ?), (?, ?)",
    params=[1, "a", 2, "b"],
    format="lazy"
)
df.show()
# -------------------------
# |"COLUMN1"  |"COLUMN2"  |
# -------------------------
# |1          |a          |
# |2          |b          |
# -------------------------


# Return a pandas DataFrame.
df = app.sql(
    query="SELECT * FROM values (?, ?), (?, ?)",
    params=[1, "a", 2, "b"],
    format="pandas"
)
print(df)
#    COLUMN1 COLUMN2
# 0        1       a
# 1        2       b

See Also

• Config