Skip to main content

​
orbitra.lake.client

​
Functions

​
get_lake_client

get_lake_client(environment: str = 'prod', credential: Optional[TokenCredential] = None) -> OrbitraLakeClient
Get the Orbitra Lake client based on the environment. Args:
  • environment: Environment to use (“prod” or “dev”). Defaults to “prod”.
  • credential: Synchronous Azure credential for API operations.
Returns:
  • Configured lake client instance.

​
Classes

​
OrbitraLakeClient

Client for interacting with the Orbitra Lake database. Methods:

​
add_column_to_table

add_column_to_table(self, namespace: str, table_name: str, column: str, column_type: AllowedColumnTypes) -> TableSchema
Add a new column to an existing table. Args:
  • namespace: The namespace where the table is located.
  • table_name: The name of the table to add the column to.
  • column: The name of the new column to add.
  • column_type: The data type of the new column.
Returns:
  • The updated schema of the table after adding the new column.
Raises:
  • LakeError: If the column is invalid, already exists or if the table does not exist.
Examples: 
from orbitra.lake import get_lake_client
client = get_lake_client()
updated_table = client.add_column_to_table(
    namespace="playground",
    table_name="my_table",
    column="new_column",
    column_type="string"
)

​
add_or_update_table

add_or_update_table(self, namespace: str, table: TableSchema, allow_column_removal: bool = False) -> TableSchema
Deprecated: Use create_or_update_table instead.

​
append_data

append_data(self, namespace: str, table_name: str, df: pd.DataFrame) -> AppendTableDataResponse
Append data to a table without overwriting existing rows. This method adds new rows to the table, leaving all existing data intact. It works for both partitioned and unpartitioned tables. Args:
  • namespace: The namespace where the table is located.
  • table_name: The name of the table to append data to.
  • df: The DataFrame containing the data to append.
Returns:
  • A response object with inserted_rows and operation_id.
Raises:
  • LakeError: If the table does not exist or if the DataFrame contains data not matching the table schema.
Examples: 
from orbitra.lake import get_lake_client
import pandas as pd

client = get_lake_client()
df = pd.DataFrame({"partition_col": [1, 2], "data_col": ["A", "B"]})
response = client.append_data(namespace="playground", table_name="my_table", df=df)

​
create_or_update_saved_query

create_or_update_saved_query(self, namespace: str, saved_query: SavedQueryCreate) -> SavedQuerySchema
Create a new saved query or update an existing one, saving old version to history. If the saved query does not exist, it is created with version 1. If it already exists, the previous definition is stored in the history directory and the version is incremented. Args:
  • namespace: The namespace where the saved query is stored.
  • saved_query: The saved query definition containing name, SQL, and optional description.
Returns:
  • The newly created or updated saved query, including its current version and timestamps.
Raises:
  • LakeError: If the SQL is invalid, not a SELECT statement, references non-existent tables, or if a conflicting table/saved query name exists when creating a new saved query.
Examples: 
from orbitra.lake import get_lake_client
from orbitra.lake.models.saved_query_schema import SavedQueryCreate

client = get_lake_client()

# First call creates the saved query
sq_def = SavedQueryCreate(name="active_clients", sql="SELECT * FROM clients WHERE is_active = TRUE")
created = client.create_or_update_saved_query("playground", sq_def)

# Second call updates the saved query SQL and bumps the version
updated_def = SavedQueryCreate(name="active_clients", sql="SELECT id, name FROM clients WHERE is_active = TRUE")
updated = client.create_or_update_saved_query("playground", updated_def)

​
create_or_update_table

create_or_update_table(self, namespace: str, table: TableSchema, allow_column_removal: bool = False, include_hash: bool = False) -> TableSchema
Create or update a table. This method will create the table if it does not exist, or update it if it does. Updates are only allowed in regular columns. Partition columns are not allowed to be updated. Args:
  • namespace: The namespace where the table is located.
  • table: The schema of the table to create or update.
  • allow_column_removal: Whether to allow column removal.
  • include_hash: Whether to include the orbitra_hash column. Defaults to False.
Returns:
  • The updated schema of the table after creating or updating it.
Raises:
  • LakeError: If there are changes in partition columns or if a column is removed and allow_column_removal is False.
Examples: 
from orbitra.lake.models.table_schema import TableSchema, ColumnSchema
from orbitra.lake import get_lake_client
client = get_lake_client()
updated_table = client.create_or_update_table(
    namespace="playground",
    table=TableSchema(
        name="my_table",
        columns=[
            ColumnSchema(name="column1", type="string", kind="regular"),
            ColumnSchema(name="column2", type="int", kind="regular")
        ]
    ),
    allow_column_removal=False
)
  • Note: In the example above, if “my_table” already exists and has a different schema, the method will remove columns not present in the new schema if allow_column_removal is set to True.

​
create_saved_query

create_saved_query(self, namespace: str, saved_query: SavedQueryCreate) -> SavedQuerySchema
Create a new saved query in the specified namespace. This helper validates the saved query SQL, ensures it references only existing tables in the given namespace, and persists the saved query definition and metadata (version, timestamps) to raw storage. Args:
  • namespace: The namespace where the saved query should be created.
  • saved_query: The saved query definition, including name, SQL text, and optional description.
Returns:
  • The created saved query with populated metadata fields such as version, created_at, and updated_at.
Raises:
  • LakeError: If the saved query name already exists as a table or saved query in the namespace, if the SQL is invalid, or if it is not a SELECT statement or references non-existent tables.
Examples: 
from orbitra.lake import get_lake_client
from orbitra.lake.models.saved_query_schema import SavedQueryCreate

client = get_lake_client()
saved_query_def = SavedQueryCreate(
    name="high_value_sales",
    sql="SELECT * FROM sales WHERE amount > 1000",
    description="Sales above R$1,000.00",
)

created = client.create_saved_query(namespace="playground", saved_query=saved_query_def)

​
create_table

create_table(self, namespace: str, table: TableSchema, include_hash: bool = False) -> TableSchema
Create a new table in the specified namespace. Args:
  • namespace: The namespace where the table should be created.
  • table: The schema of the table to create.
  • include_hash: Whether to include the orbitra_hash column. Defaults to False.
Returns:
  • The schema of the created table.
Raises:
  • LakeError: If the table already exists or if the namespace does not exist.
Examples: 
from orbitra.lake.models.table_schema import TableSchema, ColumnSchema
from orbitra.lake import get_lake_client

client = get_lake_client()
table_schema = TableSchema(
        name='my_table',
        columns=[
            ColumnSchema(name="id", type="int", kind="regular"),
            ColumnSchema(name="name", type="string", kind="regular"),
            ColumnSchema(name="value", type="double", kind="regular"),
        ]
    )

created_table = client.create_table(namespace="playground", table=table_schema)

​
delete_data

delete_data(self, namespace: str, table_name: str, partition_filters: list[PartitionFilter]) -> str
Delete data from a table based on partition filters. This method deletes data from a table based on the provided partition filters. Partition filters are combined with an AND operation. If the table has no partition columns, it deletes the entire table data. Args:
  • namespace: The namespace where the table is located.
  • table_name: The name of the table to delete data from.
  • partition_filters: A list of partition filters to apply for the delete operation. Must be empty if the table has no partition columns.
Returns:
  • An operation ID for tracking the delete operation.
Raises:
  • LakeError: If the table does not exist or if the partition filters are invalid.
Examples: 
from orbitra.lake import get_lake_client
from orbitra.lake.models import PartitionFilter

client = get_lake_client()
filter_example = [PartitionFilter(column="client", value=0, op= "<="), PartitionFilter(column="date", value="2023-01-01", op=">=")]

operation_id = client.delete_data(namespace="playground",
    table_name="my_table", partition_filters=filter_example)

​
delete_saved_query

delete_saved_query(self, namespace: str, saved_query_name: str) -> None
Delete a saved query, preserving its version history. Only the current saved query definition is removed; any historical versions previously stored remain available via list_saved_query_history and get_saved_query(..., version=...) with include_deleted=True. Args:
  • namespace: The namespace where the saved query is stored.
  • saved_query_name: The name of the saved query to delete.
Raises:
  • LakeError: If the saved query does not exist.
Examples: 
from orbitra.lake import get_lake_client

client = get_lake_client()
client.delete_saved_query(namespace="playground", saved_query_name="obsolete_query")

​
get_processed_flag

get_processed_flag(self, full_filename: str, namespace: str) -> bool
Get the processed flag for a raw file. Args:
  • full_filename: The full path and filename of the raw file to get the processed flag for.
  • namespace: Namespace used to compose the container name. The effective container is settings.orbitra_lake_raw_container_prefix + namespace.
Returns:
  • The processed flag value.
Examples: 
from orbitra.lake import get_lake_client
client = get_lake_client()
is_processed = client.get_processed_flag("test.parquet", namespace="playground")

​
get_raw_file_system

get_raw_file_system(self, namespace: str) -> AbstractFileSystem
Get a filesystem interface for the raw storage. This method returns a filesystem interface (e.g., AzureBlobFileSystem or LocalFileSystem) for direct file operations on the raw storage beyond the standard save/read operations. Args:
  • namespace: Logical namespace used to compose container/directory name.
Returns:
  • A filesystem interface for accessing raw storage.
Examples: 
from orbitra.lake import get_lake_client
client = get_lake_client()
fs = client.get_raw_file_system(namespace="playground")
# Use fsspec API for advanced operations
files = fs.ls("/")
fs.info("path/to/file.parquet")

​
get_saved_query

get_saved_query(self, namespace: str, saved_query_name: str, version: Optional[datetime] = None) -> SavedQuerySchema
Retrieve a saved query definition, optionally at a specific historical version. When version is provided, this returns the historical snapshot of the saved query at that timestamp. Otherwise, it returns the current definition. Args:
  • namespace: The namespace where the saved query is stored.
  • saved_query_name: The name of the saved query to retrieve.
  • version: Specific historical version timestamp to retrieve. If None, the latest version is returned.
Returns:
  • The requested saved query metadata and SQL definition.
Raises:
  • LakeError: If the saved query does not exist, or if the requested historical version does not exist for that saved query.
Examples: Retrieve the current version: 
from orbitra.lake import get_lake_client

client = get_lake_client()
sq = client.get_saved_query(namespace="playground", saved_query_name="simple_query")
Retrieve a specific historical version: 
from orbitra.lake import get_lake_client

client = get_lake_client()
history = client.list_saved_query_history("playground", "simple_query")
old_sq = client.get_saved_query("playground", "simple_query", version=history[0])

​
get_saved_query_data

get_saved_query_data(self, namespace: str, saved_query_name: str, scan_filters: list[Filter], limit: Optional[int] = None, engine: Literal['local', 'remote'] = 'local') -> pd.DataFrame
Execute a saved query’s SQL and return the results as a DataFrame. This method wraps the saved query SQL in a subquery, applies additional column-level filters on top, and optionally limits the number of rows returned. Args:
  • namespace: The namespace containing the saved query.
  • saved_query_name: The name of the saved query to execute.
  • scan_filters: A list of filters to apply on the resulting columns (e.g. id, date, etc.). Use an empty list to return all rows.
  • limit: Optional maximum number of rows to return. If None, all matching rows are returned.
  • engine: Query engine to use. Defaults to “local”.
Returns:
  • pd.DataFrame: The query result as a pandas DataFrame.
Raises:
  • LakeError: If the saved query does not exist, if the underlying SQL becomes invalid (for example, due to a missing table), or if filters cannot be applied.
Examples: Basic usage without filters: 
from orbitra.lake import get_lake_client

client = get_lake_client()
df = client.get_saved_query_data("playground", "my_query", scan_filters=[])
With filters and a row limit: 
from orbitra.lake import get_lake_client
from orbitra.lake.models.filter import Filter

client = get_lake_client()
filters = [Filter(column="id", value=2, op=">=")]
df = client.get_saved_query_data("playground", "my_query", scan_filters=filters, limit=10)

​
get_table_data

get_table_data(self, namespace: str, table_name: str, scan_filters: list[Filter], limit: Optional[int] = None) -> pd.DataFrame
Retrieve data from a table based on scan filters. This method retrieves data from a table based on the provided scan filters. Scan filters are combined with an AND operation. If an empty list of scan filters is provided, it retrieves all data from the table. Args:
  • namespace: The namespace where the table is located.
  • table_name: The name of the table to retrieve data from.
  • scan_filters: A list of column filters to apply for the query.
  • limit: The maximum number of rows to retrieve, defaults to None for all rows.
Returns:
  • pd.DataFrame: A DataFrame containing the data retrieved from the table.
Raises:
  • LakeError: If the table does not exist or if the scan filters are invalid.
Examples: 
from orbitra.lake import get_lake_client
from orbitra.lake.models.filter import Filter
client = get_lake_client()
df = client.get_table_data(namespace="playground", table_name="my_table", scan_filters=[Filter(column="column1", value="5", op="==")])
  • Note: If you pass scan_filters = [], it will retrieve all data from the table.

​
get_table_metadata

get_table_metadata(self, namespace: str, table_name: str) -> TableSchema
Retrieve the metadata of a table. Args:
  • namespace: The namespace where the table is located.
  • table_name: The name of the table to retrieve metadata for.
Returns:
  • The schema of the table if it exists.
Raises:
  • LakeError: If the table does not exist.
Examples: 
from orbitra.lake import get_lake_client
client = get_lake_client()
table_schema = client.get_table_metadata(namespace="playground", table_name="my_table")

​
list_namespaces

list_namespaces(self) -> list[str]
List all namespaces in the database. Returns:
  • list[str]: A list of namespace names.
Examples: 
from orbitra.lake import get_lake_client
client = get_lake_client()
namespaces = client.list_namespaces()

​
list_saved_queries

list_saved_queries(self, namespace: str) -> list[str]
List all saved queries in the specified namespace. Args:
  • namespace: The namespace to list saved queries from.
Returns:
  • list[str]: A list of saved query names available in the namespace. Returns an empty list if no saved queries are found.
Examples: 
from orbitra.lake import get_lake_client

client = get_lake_client()
names = client.list_saved_queries(namespace="playground")

​
list_saved_query_history

list_saved_query_history(self, namespace: str, saved_query_name: str, include_deleted: bool = False) -> list[datetime]
List all historical versions of a saved query. The result is a chronologically sorted list of timestamps corresponding to previous versions of the saved query. Args:
  • namespace: The namespace where the saved query is stored.
  • saved_query_name: The name of the saved query whose history should be listed.
  • include_deleted: If True, returns history even if the current saved query has been deleted. If False, raises a LakeError when the saved query does not currently exist.
Returns:
  • list[datetime]: A list of version timestamps in ascending order.
Raises:
  • LakeError: If the saved query does not exist and include_deleted is False.
Examples: 
from orbitra.lake import get_lake_client

client = get_lake_client()
history = client.list_saved_query_history("playground", "sales_summary")

​
list_tables

list_tables(self, namespace: str) -> list[str]
List all tables in the specified namespace. Args:
  • namespace: The namespace to list tables from.
Returns:
  • list[str]: A list of table names in the specified namespace.
Raises:
  • LakeError: If the namespace does not exist.
Examples: 
from orbitra.lake import get_lake_client
client = get_lake_client()
tables = client.list_tables(namespace="playground")

​
overwrite_data

overwrite_data(self, namespace: str, table_name: str, df: pd.DataFrame, check_hash: bool = False) -> OverwriteTableDataResponse
Overwrite partitions data in a table. This method overwrites the data in a table based on the partition columns present in the DataFrame. If the table has no partition columns, it overwrites the entire table data. Args:
  • namespace: The namespace where the table is located.
  • table_name: The name of the table to overwrite data in.
  • df: The DataFrame containing the data to overwrite in the table.
  • check_hash: If True, uses hash-based change detection to only overwrite data when hashes differ. Requires the table to have a hash column. Defaults to False.
Returns:
  • A response object containing information about the modified partitions and inserted rows.
Raises:
  • LakeError: If the table does not exist, if the DataFrame contains data not matching the table schema, or if check_hash is True and the table has no hash column.
Examples: 
from orbitra.lake import get_lake_client
import pandas as pd

client = get_lake_client()
df = pd.DataFrame({"partition_col": [1, 2], "data_col": ["A", "B"]})
response = client.overwrite_data(namespace="playground", table_name="my_table", df=df)

​
overwrite_data_by_custom_columns

overwrite_data_by_custom_columns(self, namespace: str, table_name: str, custom_columns: list[str], df: pd.DataFrame) -> OverwriteDataByCustomColumnsResponse
Overwrite data into a table by custom columns. This method overwrites data into a table based on the provided custom columns and data. It will delete data based on the values in the custom columns and then insert the given data frame. Args:
  • namespace: The namespace where the table is located.
  • table_name: The name of the table to overwrite data into.
  • custom_columns: A list of columns to use as custom columns.
  • df: The DataFrame containing the data to overwrite.
Returns:
  • A response object containing information about the modified custom values and inserted rows.
Raises:
  • LakeError: If the table does not exist or if the custom columns are invalid or don’t match the table schema.
Examples: Basic example with a simple custom column: 
from orbitra.lake import get_lake_client
import pandas as pd

client = get_lake_client()
df = pd.DataFrame({"custom_col": [1, 2], "data_col": ["A", "B"]})
response = client.overwrite_data_by_custom_columns(
    namespace="playground",
    table_name="my_table",
    custom_columns=["custom_col"],
    df=df
)
Example with partition column included in custom_columns: 
from orbitra.lake import get_lake_client
import pandas as pd

# For a table with 'report_date' as partition column and 'client_id' as regular column,
# include the partition column in custom_columns to overwrite specific partitions.
client = get_lake_client()
df = pd.DataFrame({
    "report_date": ["2025-01-01", "2025-01-01"],
    "client_id": ["C001", "C002"],
    "value": [100.0, 200.0]
})
response = client.overwrite_data_by_custom_columns(
    namespace="playground",
    table_name="daily_report",
    custom_columns=["report_date", "client_id"],  # partition + regular columns
    df=df
)

​
read_raw_bytes_from_blob

read_raw_bytes_from_blob(self, full_filename: str, namespace: str) -> io.BytesIO
Read a raw bytes object from the specified blob storage location. Args:
  • full_filename: The full path and filename of the raw bytes object to read from the raw storage container.
  • namespace: Namespace used to compose the container name. The effective container is settings.orbitra_lake_raw_container_prefix + namespace.
Returns:
  • io.BytesIO: The raw bytes object read from the blob storage.
Examples: 
from orbitra.lake import get_lake_client
client = get_lake_client()
bytes_io = client.read_raw_bytes_from_blob("test.parquet", namespace="playground")

​
read_raw_df_from_blob

read_raw_df_from_blob(self, full_filename: str, namespace: str) -> pd.DataFrame
Reads a raw Parquet file from the specified blob storage location and returns its contents as a pandas DataFrame. Args:
  • full_filename: The full path and filename of the Parquet file to read from the raw storage container.
  • namespace: Namespace used to compose the container name. The effective container is settings.orbitra_lake_raw_container_prefix + namespace.
Returns:
  • pd.DataFrame: The contents of the Parquet file as a pandas DataFrame.
Examples: 
from orbitra.lake import get_lake_client
client = get_lake_client()
df = client.read_raw_df_from_blob("test.parquet", namespace="playground")

​
remove_column_from_table

remove_column_from_table(self, namespace: str, table_name: str, column: str) -> TableSchema
Remove a column from an existing table. Args:
  • namespace: The namespace where the table is located.
  • table_name: The name of the table to remove the column from.
  • column: The name of the column to remove.
Returns:
  • The updated schema of the table after removing the column.
Raises:
  • LakeError: If the table or column does not exist or if it is a reserved column.
Examples: 
from orbitra.lake import get_lake_client
client = get_lake_client()
updated_table = client.remove_column_from_table(
    namespace="playground",
    table_name="my_table",
    column="old_column"
)

​
run_query

run_query(self, namespace: str, query: str, engine: Literal['local', 'remote'] = 'local') -> pd.DataFrame
Run a query using the selected environment as query engine. Args:
  • namespace: The namespace to run the query in.
  • query: The query to run.
  • engine: The engine to use for the query.
Returns:
  • pd.DataFrame: A DataFrame containing the retrieved data.
Raises:
  • LakeError: If the query is invalid or if the engine is not supported.
Examples: 
from orbitra.lake import get_lake_client
client = get_lake_client()
query = "SELECT * FROM my_table WHERE column1 = 'value1'"
df = client.run_query(namespace="playground", query=query, engine="local")

​
save_raw_bytes_to_blob

save_raw_bytes_to_blob(self, bytes_io: io.BytesIO, full_filename: str, namespace: str) -> bool
Save a bytes object as a raw blob in Azure Blob Storage. Args:
  • bytes_io: The bytes object to persist.
  • full_filename: The blob path, including virtual directories, e.g. "finance/2025/09/transactions.parquet".
  • namespace: Namespace used to compose the container name. The effective container is settings.orbitra_lake_raw_container_prefix + namespace.
Returns:
  • True if the bytes object was stored, False if it already exists and is the same.
Examples: 
from orbitra.lake import get_lake_client
import io
bytes_io = io.BytesIO(b"Hello, world!")
client = get_lake_client()
result = client.save_raw_bytes_to_blob(bytes_io, "test.txt", namespace="playground")

​
save_raw_df_to_blob

save_raw_df_to_blob(self, df: pd.DataFrame, full_filename: str, namespace: str) -> bool
Save a DataFrame as a Parquet blob in Azure Blob Storage. The DataFrame is serialized to Parquet in memory and uploaded to the configured storage account and container. Existing blobs will be overwritten. Args:
  • df: The DataFrame to persist.
  • full_filename: The blob path, including virtual directories, e.g. "finance/2025/09/transactions.parquet".
  • namespace: Namespace used to compose the container name. The effective container is settings.orbitra_lake_raw_container_prefix + namespace.
Returns:
  • True if the DataFrame was stored, False if it already exists and is the same.
Examples: 
from orbitra.lake import get_lake_client
import pandas as pd
df = pd.DataFrame({"id": [1, 2], "value": ["A", "B"]})
client = get_lake_client()
result = client.save_raw_df_to_blob(df, "test.parquet", namespace="playground")

​
set_processed_flag

set_processed_flag(self, full_filename: str, namespace: str, is_processed: bool) -> None
Set the processed flag for a raw file. Args:
  • full_filename: The full path and filename of the raw file to set the processed flag for.
  • namespace: Namespace used to compose the container name. The effective container is settings.orbitra_lake_raw_container_prefix + namespace.
  • is_processed: The processed flag value to set.
Examples: 
from orbitra.lake import get_lake_client
client = get_lake_client()
client.set_processed_flag("test.parquet", namespace="playground", is_processed=True)