Store¶

The Store interface provides access to platform data and resources within your apps.

StoreInterface¶

StoreInterface ¶

StoreInterface(flow_id: str, user: Optional[str] = None, step_name: Optional[str] = None, is_action: bool = False, bucket_name: Optional[str] = None)

The StoreInterface class is the main interface to storing and retrieving metadata. It provides convenience methods for storing input data, assets and flow metadata and also methods for retrieving previously saved data.

EXAMPLE:

# Imports from virtualitics_sdk import StoreInterface...
class ExampleStep(Step): def run(self, flow_metadata):...
future_dropdown = Dropdown( ["Rows", "Columns"], selected=["Rows"], title="Dashboard Options", label="Option Selector", ) store_interface.create_future_element(future_dropdown, future_step.name)  table_links = [ "https://www.google.com", store_interface.create_element_link(future_dropdown, future_step.name) ]  table = Table(example_dataset, title="Example Table", description="This is a table showing cells/text color", downloadable=True, links=table_links)...
class FutureSte(Step): def run(self, flow_metadata):...
dropdown = store_interface.get_element(example_step.name, "Dashboard Options")...

create_element_link ¶

create_element_link(element: Element, step_name: Optional[str] = None)

Create a link to any element present in the current step or in the previous ones.

Automatically includes navigation metadata for elements inside SegmentedControls, Accordions and DASHBOARD steps (tab switching).

Parameters:

element (Element) –

Element that we want to link.
step_name (Optional[str], default: None ) –

The name of the step the element was in.

Returns:

–

EXAMPLE:

# Imports from virtualitics_sdk import StoreInterface...
class PreviousStep(Step): def run(self, flow_metadata):...
text = TextInput(title="Some Text")...
class ExampleStep(Step): def run(self, flow_metadata):...
prev_text_input = store_interface.get_element(previous_step.name, "Some Text")  table_links = [ "https://www.google.com", store_interface.create_element_link(prev_text_input, previous_step.name) ]  table = Table(example_dataset, title="Example Table", description="This is a table showing links", downloadable=True, links=table_links)...

save_asset ¶

save_asset(asset: Asset, overwrite: bool = False, asset_id: Optional[str] = None, serialization_method: Optional[AssetPersistenceMethod] = None)

Save an Asset. This is useful for storing objects, datasets, models to be used in other apps or within the current flow. Assets are persisted until they are deleted (even if the flow they were created in is deleted)

Parameters:

asset (Asset) –

The asset object to save.
overwrite (bool, default: False ) –

bool: Overwrite the existing asset with the same label and type if it exists
asset_id (Optional[str], default: None ) –

str: An optional asset_id to use when overwriting
serialization_method (Optional[AssetPersistenceMethod], default: None ) –

AssetPersistenceMethod: An Optional argument to force serialization using a specific method

get_asset ¶

get_asset(label: Optional[str] = None, type: Optional[AssetType] = None, name: Optional[str] = None, time_created: Optional[str] = None, asset_id: Optional[str] = None) -> Asset

Retrieve a saved Asset. This function returns (at most) 1 asset, use get_assets for retrieving a list of assets that matches the supplied argument values. This function will only return an asset that the requesting user has access to

Parameters:

label (Optional[str], default: None ) –

The label of the Asset, defaults to None.
type (Optional[AssetType], default: None ) –

The type of asset, defaults to None.
name (Optional[str], default: None ) –

The name for the asset, defaults to None.
time_created (Optional[str], default: None ) –

The time the asset was created. This is especially optional and only necessary when you want to receive an Asset by timestamp as well as other metadata, defaults to None.
asset_id (Optional[str], default: None ) –

The unique identifier of a specific asset, defaults to None

Returns:

Asset –

The Asset object.

Raises:

ValueError –

If the asset label and type are both None.

get_assets ¶

get_assets(label: Optional[str] = None, type: Optional[AssetType] = None, name: Optional[str] = None, asset_id: Optional[str] = None) -> List[Asset]

Retrieve multiple saved Assets. Providing any of the attributes will filter all available assets to retrieve only the ones which match the given label, type, name, combination. Providing none of these descriptors will retrieve all available assets.

Parameters:

label (Optional[str], default: None ) –

The label of the Asset, defaults to None.
type (Optional[AssetType], default: None ) –

The type of asset, defaults to None.
name (Optional[str], default: None ) –

The name for the asset, defaults to None.
asset_id (Optional[str], default: None ) –

The unique identifier of a specific asset, defaults to None

Returns:

List[Asset] –

List of Asset objects.

get_asset_by_id ¶

get_asset_by_id(asset_id: str) -> Asset

Retrieve a saved asset using the asset_id

Parameters:

asset_id (str) –

The unique identifier of a specific asset.

Returns:

Asset –

The Asset object.

get_model ¶

get_model(label: Optional[str] = None, name: Optional[str] = None) -> Model

This is a convenience method for getting Assets that have a "Model" type

Parameters:

label (Optional[str], default: None ) –

The label of the asset, defaults to None.
name (Optional[str], default: None ) –

The name of the asset, defaults to None.

Returns:

Model –

The Model asset.

get_dataset ¶

get_dataset(label: Optional[str] = None, name: Optional[str] = None) -> Dataset

This is a convenience method for getting Assets that have a "Dataset" type

Parameters:

label (Optional[str], default: None ) –

The label of the asset, defaults to None.
name (Optional[str], default: None ) –

The name of the asset, defaults to None.

Returns:

Dataset –

The Dataset asset.

get_schema ¶

get_schema(label: Optional[str] = None, name: Optional[str] = None) -> Schema

This is a convenience method for getting Assets that have a "Schema" type

Parameters:

label (Optional[str], default: None ) –

The label of the asset, defaults to None.
name (Optional[str], default: None ) –

The name of the asset, defaults to None.

Returns:

Schema –

The Schema asset.

get_s3_asset ¶

get_s3_asset(path: str)

Retrieve an asset from a pre-specified bucket. In order to use, please initialize the store interface with the bucket_name parameter. TODO: this asset wont exist in the asset store, this function might need to be removed if possible

Parameters:

path (str) –

The path to the asset in the s3 bucket.

Returns:

–

returns the asset from s3

update_page_from_live_card ¶

update_page_from_live_card(section_title: str, card_id: str, **step_clients)

:meta private:

Parameters:

section_title (str) –
card_id (str) –
step_clients –

Returns:

get_current_step_user_input ¶

get_current_step_user_input(data_source_reference_id: str = '', **kwargs) -> BytesIO

Get the raw bytes that were uploaded in the current step (prior to the step action). This can be useful for doing data validation on the uploaded data in a dynamic page update function. When uploading data using the DataSource element the data is not converted into a dataframe until the step action is run (Next button) which also puts the data on the subsequent steps in-link. This means that you cannot access the uploaded object using the common methods which retrieve data from the in-link

Parameters:

data_source_reference_id (str, default: '' ) –

the eference_id of the DataSource element

Returns:

BytesIO –

a BytesIO object of the data that was uploaded

db_to_pandas ¶

db_to_pandas(query: str, conn_name: str, connection_owner: Optional[str] = None, **kwargs)

NOTICE: As of version 1.23.0 this function is depreciated

Given a SQL query and a connection name of a connection stored in the connection store execute the query against the defined data store connection and return the result set as a pandas data frame

Parameters:

query (str) –

The SQL query to execute against the supplied data store
conn_name (str) –

The connection name where database credentials, host, etc will be retrieved from the connection store
connection_owner (Optional[str], default: None ) –

(optional) The owner of the connection being retrieved, this defaults to the current user
kwargs –

additional keyword arguments, for databricks connections http_path can be supplied here to override the default http_path stored in the connection store

Returns:

–

A pandas data frame

pandas_to_db ¶

pandas_to_db(_df: DataFrame, table: str, conn_name: str, connection_owner: Optional[str] = None, if_exists: str = 'fail', **kwargs)

NOTICE: As of version 1.23.0 this function is depreciated

Write the contents of a dataframe to the supplied data store table. Retrieve DB connection details by supplying the connection name and the connection owner (optional)

Parameters:

_df (DataFrame) –

The pandas dataframe to be written
table (str) –

The destination table where data will be written
conn_name (str) –

The connection name where database credentials, host, etc will be retrieved from the connection store
connection_owner (Optional[str], default: None ) –

(optional) The owner of the connection being retrieved, this defaults to the current user
if_exists (str, default: 'fail' ) –

What to do if the table already exists: 'fail', 'replace', 'append'
kwargs –

additional keyword arguments, for databricks connections http_path can be supplied here to override the default http_path stored in the connection store

Returns:

async_reset_app `async` ¶

async_reset_app() -> None

Reset the app back to its initial state and re-run the first step.

Async version for use in trigger callbacks or other async contexts. This is the programmatic equivalent of pressing the Reset button in a singleton app. All step data is wiped, pages are recreated from the app definition, and the first step is re-executed.

reset_app ¶

reset_app() -> None

Reset the app back to its initial state and re-run the first step.

This is the programmatic equivalent of pressing the Reset button in a singleton app. All step data is wiped, pages are recreated from the app definition, and the first step is re-executed.

Use :meth:async_reset_app in async contexts (e.g. trigger callbacks).

list_fixture_data ¶

list_fixture_data(prefix: Optional[str] = None) -> List[str]

List all of the objects stored within the deployment's fixture path in s3://{meta-data-bucket}/fixture/{prefix}

Parameters:

prefix (Optional[str], default: None ) –

Optional prefix to filter objects

Returns:

List[str] –

list of s3 keys

str_2_connection_type `staticmethod` ¶

str_2_connection_type(list_connection_type: List[str]) -> List[ConnectionType]

Converts a list of strings representing the connection type to a list of ConnectionType

Parameters:

list_connection_type (List[str]) –

list of strings representing the connection type

Returns:

List[ConnectionType] –

list of ConnectionType

get_boto3_s3_client_from_connection_store ¶

get_boto3_s3_client_from_connection_store(connection_id: str, **kwargs) -> boto3.client

Using a connection stored in the connection store, create and return a boto3 client configured with the credentials stored in the connection store

Parameters:

connection_id (str) –

the UID of a connection stored in the connection store
kwargs –

additional keyword arguments to pass to the boto3.Session or boto3.client objects

Returns:

client –

a boto3.client('s3')

save_datastore_asset ¶

save_datastore_asset(data: DataFrame, name: str, asset_id: Optional[str] = None, description: Optional[str] = '', overwrite_if_exists: bool = True, encode_columns: Optional[List] = None, indexes: Optional[DatasetIndex] = None)

Write a pandas dataframe to a postgres table, and create an asset record. This allows for more efficient querying of the underlying data for certain use cases. Instead of being required to read the entire dataset into a dataframe in memory and perform transform, filter, select, etc. operations on the data. Instead this enables use of the query_datastore_asset function which allows for those operations to be executed in the db returning a smaller result set

Parameters:

data (DataFrame) –

a pandas dataframe containing all the data to be written
name (str) –

a name for this dataset, this should be a unique identifier that refers to the dataset
asset_id (Optional[str], default: None ) –

if overwriting an existing datastore asset, providing the asset_id specifies which asset will be replaced
description (Optional[str], default: '' ) –

a description of the asset, displayed in the Assets page
overwrite_if_exists (bool, default: True ) –

overwrite the existing data with this name?
indexes (Optional[DatasetIndex], default: None ) –

an optional list of indexes to specify eg. [{'columns': ['column1'], 'unique': True}]

Returns:

–

EXAMPLE:

import seaborn as sns  from virtualitics_sdk import StoreInterface   store_interface = StoreInterface(**flow_metadata) data = sns.load_dataset("iris") asset_id = store_interface.save_datastore_asset( data=data, name="iris", indexes=[{"columns": ["sepal_length"], "unique": False}] )

query_datastore_asset ¶

query_datastore_asset(model: Type[Union[MappedAsDataclass, DeclarativeBase]], select_: Union[List[ColumnElement], None] = None, where_: Union[List[ColumnElement], None] = None, name: Optional[str] = None, asset_id: Optional[str] = None) -> pandas.DataFrame

Query a previously saved datastore asset. Provide a SQLAlchemy BaseModel that describes the table where the asset is stored and optional select and where clauses. Returns a pandas dataframe that represents the ResultSet.

Parameters:

model (Type[Union[MappedAsDataclass, DeclarativeBase]]) –

A sqlalchemy Base model
select_ (Union[List[ColumnElement], None], default: None ) –

a list of Column Expressions, any valid sqlalchemy column expression is acceptable, including sqlalchemy.func expressions
where_ (Union[List[ColumnElement], None], default: None ) –

a list of Column Expressions to filter the dataset, any valid sqlalchemy column expression that resolves to a boolean value is acceptable, including sqlalchemy.func expressions
name (Optional[str], default: None ) –

the dataset asset name (either the asset name or the asset_id are required)
asset_id (Optional[str], default: None ) –

the dataset asset identifier (either the asset name or the asset_id are required)

Returns:

DataFrame –

EXAMPLE:

from sqlalchemy.orm import DeclarativeBase, Mapped, MappedAsDataclass, mapped_column   class Base(MappedAsDataclass, DeclarativeBase): pass   class Iris(Base): <strong>tablename</strong> = "iris"  id: Mapped[int] = mapped_column(sa.BigInteger, primary_key=True) sepal_length: Mapped[Optional[float]] = mapped_column(sa.Double(53)) sepal_width: Mapped[Optional[float]] = mapped_column(sa.Double(53)) petal_length: Mapped[Optional[float]] = mapped_column(sa.Double(53)) petal_width: Mapped[Optional[float]] = mapped_column(sa.Double(53)) species: Mapped[Optional[str]] = mapped_column(sa.Text)   store_interface = StoreInterface(**flow_metadata) df = store_interface.query_datastore_asset( model=Iris, name="iris", select_=[Iris.sepal_length, Iris.sepal_width], where_=[Iris.sepal_length > 0], )

get_current_user_details ¶

get_current_user_details() -> tuple[str, list[str]]

Retrieve the details of the currently authenticated user.

This function returns a tuple containing:
- A string indicating the user's role.
- A list of strings representing the names of the groups the user belongs to.

Returns:

tuple[str, list[str]] –

tuple[str, list[str]]: A tuple with the user's role and a list of associated group names.

get_custom_attributes ¶

get_custom_attributes() -> dict

Retrieve custom attributes for the currently authenticated user.

This method returns the custom_attributes stored for the user, which may include DoD-specific attributes like DoD ID from SAML assertions.

Returns:

dict –

dict: The user's custom attributes, or empty dict if none exist.

record_audit_log ¶

record_audit_log(event: AuditLogEvent, resource_id: Optional[str] = None, **extra_detail: Any) -> None

Record an audit event (fire-and-forget). User context is automatically extracted from the current flow.

Do not pass credentials, tokens, or PII as extra_detail.

Parameters:

event (AuditLogEvent) –

An AuditLogEvent enum value. Use AuditLogEvent.EXTERNAL_API_CALL for external API interactions.
resource_id (Optional[str], default: None ) –

Optional resource identifier (e.g., asset_id, external record ID).

extra_detail (Any, default: {} ) –

EXAMPLE:

store_interface.record_audit_log( AuditLogEvent.READ_ASSET, resource_id="asset_123", )  store_interface.record_audit_log( AuditLogEvent.EXTERNAL_API_CALL, resource_id="salesforce_contact_456", api_endpoint="https://api.salesforce.com/contacts", operation="update", records_affected=1, )

record_audit_log_async `async` ¶

record_audit_log_async(event: AuditLogEvent, resource_id: Optional[str] = None, **extra_detail: Any) -> None

Record an audit event (async/awaitable). User context is automatically extracted from the current flow.

Do not pass credentials, tokens, or PII as extra_detail.

Parameters:

event (AuditLogEvent) –

An AuditLogEvent enum value. Use AuditLogEvent.EXTERNAL_API_CALL for external API interactions.
resource_id (Optional[str], default: None ) –

Optional resource identifier (e.g., asset_id, external record ID).

extra_detail (Any, default: {} ) –

EXAMPLE:

await store_interface.record_audit_log_async( AuditLogEvent.EXTERNAL_API_CALL, resource_id="external_dataset_789", api_endpoint="https://api.example.com/data", rows_fetched=1000, )

Store¶

StoreInterface¶

StoreInterface ¶

create_element_link ¶

save_asset ¶

get_asset ¶

get_assets ¶

get_asset_by_id ¶

get_model ¶

get_dataset ¶

get_schema ¶

get_s3_asset ¶

update_page_from_live_card ¶

get_current_step_user_input ¶

db_to_pandas ¶

pandas_to_db ¶

async_reset_app `async` ¶

reset_app ¶

list_fixture_data ¶

str_2_connection_type `staticmethod` ¶

get_boto3_s3_client_from_connection_store ¶

save_datastore_asset ¶

query_datastore_asset ¶

get_current_user_details ¶

get_custom_attributes ¶

record_audit_log ¶

record_audit_log_async `async` ¶

Overview¶

Basic Usage¶

Common Operations¶

Getting Datasets¶

User Preferences¶

App Metadata¶

DrilldownStoreInterface¶

DrilldownStoreInterface ¶

update_progress `staticmethod` ¶

aupdate_progress `async` `staticmethod` ¶

Best Practices¶

See Also¶

Store¶

StoreInterface¶

StoreInterface ¶

create_element_link ¶

save_asset ¶

get_asset ¶

get_assets ¶

get_asset_by_id ¶

get_model ¶

get_dataset ¶

get_schema ¶

get_s3_asset ¶

update_page_from_live_card ¶

get_current_step_user_input ¶

db_to_pandas ¶

pandas_to_db ¶

async_reset_app async ¶

reset_app ¶

list_fixture_data ¶

str_2_connection_type staticmethod ¶

get_boto3_s3_client_from_connection_store ¶

save_datastore_asset ¶

query_datastore_asset ¶

get_current_user_details ¶

get_custom_attributes ¶

record_audit_log ¶

record_audit_log_async async ¶

Overview¶

Basic Usage¶

Common Operations¶

Getting Datasets¶

User Preferences¶

App Metadata¶

DrilldownStoreInterface¶

DrilldownStoreInterface ¶

update_progress staticmethod ¶

aupdate_progress async staticmethod ¶

Best Practices¶

See Also¶

async_reset_app `async` ¶

str_2_connection_type `staticmethod` ¶

record_audit_log_async `async` ¶

update_progress `staticmethod` ¶

aupdate_progress `async` `staticmethod` ¶