# Databricks Apps Cookbook

> Ready-to-use code snippets for building data and AI applications using Databricks Apps

This file contains all documentation content in a single document following the llmstxt.org standard.

## Connect an MCP server


This recipe connects to an [MCP](https://modelcontextprotocol.io/overview) server for AI applications using GitHub as an example and the Unity Catalog [HTTP connection](https://docs.databricks.com/aws/en/query-federation/http) for secure and governed access.

## Code snippets

### OAuth User to Machine Per User (On-behalf-of-user)

```python title="app.py"
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ExternalFunctionRequestHttpMethod

from flask import request

token = request.headers.get("x-forwarded-access-token")
w = WorkspaceClient(token=token, auth_type="pat")

def init_mcp_session(w: WorkspaceClient, connection_name: str):
    init_payload = {
        "jsonrpc": "2.0",
        "id": "init-1",
        "method": "initialize",
        "params": {}
    }
    response = w.serving_endpoints.http_request(
        conn=connection_name,
        method=ExternalFunctionRequestHttpMethod.POST,
        path="/",
        json=init_payload,
    )
    return response.headers.get("mcp-session-id")

connection_name = "github_u2m_connection"
http_method = ExternalFunctionRequestHttpMethod.POST
path = "/"
headers = {"Content-Type": "application/json"}
payload = {"jsonrpc": "2.0", "id": "list-1", "method": "tools/list"}

session_id = init_mcp_session(w, connection_name)
headers["Mcp-Session-Id"] = session_id

response = w.serving_endpoints.http_request(
    conn=connection_name,
    method=http_method,
    path=path,
    headers=headers,
    json=payload,
)
print(response.json())
```

### Bearer token

```python title="app.py"
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ExternalFunctionRequestHttpMethod

w = WorkspaceClient()

response = w.serving_endpoints.http_request(
    conn="github_u2m_connection",
    method=ExternalFunctionRequestHttpMethod.GET,
    path="/traffic/views",
    headers={"Accept": "application/vnd.github+json"},
    json={
        "jsonrpc": "2.0",
        "id": "init-1",
        "method": "initialize",
        "params": {}
    },
)

print(response.json())
```

## Resources

- [Unity Catalog HTTP Connection](https://docs.databricks.com/aws/en/query-federation/http) with the MCP (/mcp) base path

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CONNECTION` permission on the HTTP Connection

When using OAuth User to Machine Per User (On-behalf-of-user), you need to configure [User authorization](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/auth#user-authorization) by adding the Unity Catalog connection or other scopes.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Dash](https://pypi.org/project/dash/) - `dash`
- [Uvicorn](https://pypi.org/project/uvicorn/) - `uvicorn`
- [MCP CLI](https://pypi.org/project/mcp/) - `mcp[cli]`

```python title="requirements.txt"
databricks-sdk
dash
uvicorn
mcp[cli]
```

---

## Invoke a model


This recipe invokes a model hosted on [Mosaic AI Model Serving](https://docs.databricks.com/aws/en/machine-learning/model-serving/) using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/) and returns the result. Choose either a traditional ML model or a large language model (LLM).

## Code snippets

### Traditional Machine Learning

#### Using `dataframe_split` (JSON-serialized DataFrame in split orientation)

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="custom-regression-model",
    dataframe_split={
        "columns": ["feature1", "feature2"],
        "data": [[1.5, 2.5]]
    }
)
```

#### Using `dataframe_records` (JSON-serialized DataFrame in records orientation)

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="custom-regression-model",
    dataframe_records={
        "feature1": [1.5],
        "feature2": [2.5]
    }
)
```

#### Using `instances` (Tensor inputs in row format for TensorFlow/PyTorch models)

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

tensor_input = [[1.0, 2.0, 3.0]]
response = w.serving_endpoints.query(
    name="tensor-processing-model",
    instances=tensor_input,
)
```

#### Using `inputs` (Tensor inputs in columnar format for TensorFlow/PyTorch models)

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

tensor_input = {
    "input1": [1.0, 2.0, 3.0],
    "input2": [4.0, 5.0, 6.0],
}
response = w.serving_endpoints.query(
    name="tensor-processing-model",
    inputs=tensor_input,
)
```

### Large language models (LLMs)

#### Using `prompt` (Input text for completion tasks)

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="llm-text-completions-model",
    prompt="Generate a recipe for building scalable Databricks Apps.",
    temperature=0.5,
)
```

#### Using `messages` (List of chat messages for conversational models)

```python title="app.py"
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ChatMessage, ChatMessageRole

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="chat-assistant-model",
    messages=[
        ChatMessage(
            role=ChatMessageRole.SYSTEM,
            content="You are a helpful assistant.",
        ),
        ChatMessage(
            role=ChatMessageRole.USER,
            content="Provide tips for deploying Databricks Apps.",
        ),
    ],
)
```

#### Using `input` (Input text for embedding tasks)

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="embedding-model",
    input="Databricks provides a unified analytics platform.",
)
```

## Resources

- [Model Serving endpoint](https://docs.databricks.com/aws/en/machine-learning/model-serving/manage-serving-endpoints)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN QUERY` on the model serving endpoint

See [Manage permissions on your model serving endpoint](https://docs.databricks.com/aws/en/machine-learning/model-serving/manage-serving-endpoints#manage-permissions-on-your-model-serving-endpoint) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Dash](https://pypi.org/project/dash/) - `dash`

```python title="requirements.txt"
databricks-sdk
dash
```

---

## Run vector search


This recipe performs a vector search query on a [Mosaic AI Vector Search](https://docs.databricks.com/aws/en/generative-ai/vector-search) index using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/).

## Code snippet

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()
openai_client = w.serving_endpoints.get_open_ai_client()

EMBEDDING_MODEL_ENDPOINT_NAME = "databricks-gte-large-en"

def get_embeddings(text):
    try:
        response = openai_client.embeddings.create(
            model=EMBEDDING_MODEL_ENDPOINT_NAME, input=text
        )
        return response.data[0].embedding
    except Exception as e:
        return f"Error generating embeddings: {e}"

def run_vector_search(prompt: str) -> str:
    prompt_vector = get_embeddings(prompt)
    if prompt_vector is None or isinstance(prompt_vector, str):
        return f"Failed to generate embeddings: {prompt_vector}"

    columns_to_fetch = [col.strip() for col in columns.split(",") if col.strip()]

    try:
        query_result = w.vector_search_indexes.query_index(
            index_name=index_name,
            columns=columns_to_fetch,
            query_vector=prompt_vector,
            num_results=3,
        )
        return query_result.result.data_array
    except Exception as e:
        return f"Error during vector search: {e}"
```

## Resources

- [Vector Search endpoint](https://docs.databricks.com/aws/en/generative-ai/vector-search)
- [Vector Search index](https://docs.databricks.com/aws/en/generative-ai/vector-search)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CATALOG` on the catalog that contains the Vector Search index
- `USE SCHEMA` on the schema that contains the Vector Search index
- `SELECT` on the Vector Search index

See [Query a vector search endpoint](https://docs.databricks.com/aws/en/generative-ai/vector-search) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Dash](https://pypi.org/project/dash/) - `dash`

```python title="requirements.txt"
databricks-sdk
dash
```

---

## Get current user


This recipe gets information about the user accessing this Databricks App from their [HTTP headers](https://docs.databricks.com/en/dev-tools/databricks-apps/app-development.html#what-http-headers-are-passed-to-databricks-apps).

## Code snippet

```python title="app.py"
from flask import request

headers = request.headers
email = headers.get("X-Forwarded-Email")
username = headers.get("X-Forwarded-Preferred-Username")
user = headers.get("X-Forwarded-User")
ip = headers.get("X-Real-Ip")
print(f"E-mail: {email}, username: {username}, user: {user}, ip: {ip}")
```

:::info
This sample requires on-behalf-of-user authentication to be enabled for your app to access the X-Forwarded-Access-Token header. Without this, you will only have access to basic user information from the headers, not the detailed information from the Databricks API. Without the user token present, w.current_user.me() will return information about the app service principal.
:::

## Permissions

No permissions configuration required.

## Dependencies

- [Dash](https://pypi.org/project/dash/) - `dash`

```python title="requirements.txt"
dash
```

---

## Embed a dashboard


This recipe embeds a [Databricks AI/BI dashboard](https://docs.databricks.com/aws/en/dashboards/) into a Databricks App.

## Code snippet

```python title="app.py"
from dash import html

iframe_source = "https://workspace.azuredatabricks.net/embed/dashboardsv3/dashboard-id"

html.Iframe(
    src=iframe_source,
    width="700px",
    height="600px",
    style={"border": "none"}
)
```

:::info

Copy and paste the dashoard embedding URL from the dashboard UI **Share** -> **Embed iframe**.

:::

## Resources

- [AI/BI dashboard](https://docs.databricks.com/aws/en/dashboards/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN VIEW` permission on the dashboard

:::info

A workspace admin needs to enable dashboard embedding in the Security settings of your Databricks workspace for specific domains (e.g., databricksapps.com) or all domains for this sample to work.

:::

## Dependencies

- [Dash](https://pypi.org/project/dash/) - `dash`

```python title="requirements.txt"
dash
```

---

## Chat with a Genie Space


This app uses the [AI/BI Genie](https://www.databricks.com/product/ai-bi) [Conversations API](https://docs.databricks.com/api/workspace/genie) to let users ask questions about your data for instant insights (answers and table-like output). Visualizations aren't yet supported in the API.

## Code snippet

Refer to the Dash Cookbook Genie source code for the full implementation.

```python title="app.py"

from databricks.sdk import WorkspaceClient

def get_query_result(statement_id):
    # For simplicity, let's say data fits in one chunk, query.manifest.total_chunk_count = 1

    result = w.statement_execution.get_statement(statement_id)
    return pd.DataFrame(
        result.result.data_array, columns=[i.name for i in result.manifest.schema.columns]
    )

def process_genie_response(response):
    for i in response.attachments:
        if i.text:
            print(f"A: {i.text.content}")
        elif i.query:
            data = get_query_result(response.query_result.statement_id)
            print(f"A: {i.query.description}")
            print(f"Data: {data}")
            print(f"Generated code: {i.query.query}")
            

# Configuration
w = WorkspaceClient()
genie_space_id = "01f0023d28a71e599b5a62f4117516d4"

prompt = "Ask a question..."
follow_up_prompt = "Ask a follow-up..."

# Start the conversation
conversation = w.genie.start_conversation_and_wait(genie_space_id, prompt)
process_genie_response(conversation)

# Continue the conversation
follow_up_conversation = w.genie.create_message_and_wait(
    genie_space_id, conversation.conversation_id, follow_up_prompt
)
process_genie_response(follow_up_conversation)
```

:::info

Copy and paste the Genie space ID from the Genie UI URL as rooms/SPACE-ID?o=.

:::

## Resources

- [Genie](https://www.databricks.com/what-aibi-genie)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `SELECT` on the Unity Catalog table
- `CAN USE` the SQL warehouse
- `CAN VIEW` the Genie Space

## Dependencies

- [Dash](https://pypi.org/project/dash/) - `dash`
- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Pandas](https://pypi.org/project/pandas/) - `pandas`

```python title="requirements.txt"
dash
databricks-sdk
pandas
```

---

## Connect to a cluster


This recipe uses [Databricks Connect](https://docs.databricks.com/en/dev-tools/databricks-connect/python/index.html) to execute pre-defined Python or SQL code on a **shared** cluster with UI inputs.

## Code snippet

```python title="app.py"

from databricks.connect import DatabricksSession

cluster_id = "0709-132523-cnhxf2p6"

spark = DatabricksSession.builder.remote(
    host=os.getenv("DATABRICKS_HOST"),
    cluster_id=cluster_id
).getOrCreate()

# SQL operations example
a = "(VALUES (1, 'A1'), (2, 'A2'), (3, 'A3')) AS a(id, value)"
b = "(VALUES (2, 'B1'), (3, 'B2'), (4, 'B3')) AS b(id, value)"

# Inner join example
query = f"SELECT a.id, a.value AS value_a, b.value AS value_b FROM {a} INNER JOIN {b} ON a.id = b.id"
result = spark.sql(query).toPandas()
print(result)

# Generate sequence
result = spark.range(10).toPandas()
print(result)
```

:::info

You also have the option to [connect to serverless compute using Databricks Connect](https://docs.databricks.com/aws/en/compute/serverless/).

:::

## Resources

- [All-purpose compute](https://docs.databricks.com/aws/en/compute/use-compute) or [serverless compute](https://docs.databricks.com/aws/en/compute/serverless/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN ATTACH TO` permission on the cluster

See [Compute permissions](https://docs.databricks.com/aws/en/compute/clusters-manage#compute-permissions) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Dash](https://pypi.org/project/dash/) - `dash`

```python title="requirements.txt"
databricks-sdk
dash
```

---

## External connections


This recipe demonstrates how to use Unity Catalog-managed external [HTTP connections](https://docs.databricks.com/aws/en/query-federation/http) for secure and governed access to MCP and non-MCP servers, for example, to GitHub, or Jira, and Slack.

## Code snippets

### OAuth User to Machine Per User (On-behalf-of-user)

```python
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ExternalFunctionRequestHttpMethod
from flask import request

token = request.headers.get("x-forwarded-access-token")
w = WorkspaceClient(token=token, auth_type="pat")

response = w.serving_endpoints.http_request(
    conn="github_u2m",
    method=ExternalFunctionRequestHttpMethod.GET,
    path="/user",
    headers={"Accept": "application/vnd.github+json"},
)

print(response.json())
```

### Bearer token

```python title="app.py"
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ExternalFunctionRequestHttpMethod
w = WorkspaceClient()

response = w.serving_endpoints.http_request(
    conn="github_connection",
    method=ExternalFunctionRequestHttpMethod.GET,
    path="/traffic/views",
    headers={"Accept": "application/vnd.github+json"},
)

print(response.json())
```

## Resources

- [Unity Catalog HTTP Connection](https://docs.databricks.com/aws/en/query-federation/http)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CONNECTION` permission on the HTTP Connection

When using OAuth User to Machine Per User (On-behalf-of-user), you need to configure [User authorization](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/auth#user-authorization) by adding the Unity Catalog connection or other scopes.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Dash](https://pypi.org/project/dash/) - `dash`

```python title="requirements.txt"
databricks-sdk
dash
```

---

## Retrieve a secret


This recipe retrieves a [Databricks secret](https://docs.databricks.com/en/security/secrets/index.html). Use secrets to securely connect to external services and APIs.

## Code snippet

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

def get_secret(scope, key):
    try:
        secret_response = w.secrets.get_secret(scope=scope, key=key)
        decoded_secret = base64.b64decode(secret_response.value).decode('utf-8')
        return decoded_secret
    except Exception as e:
        print("Secret not found or inaccessible")

scope_name = "my_secret_scope"
secret_key = "api_key"
secret = get_secret(scope_name, secret_key)
```

## Resources

- [Secret scope and secret](https://docs.databricks.com/aws/en/security/secrets/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN READ` on the secret scope

See [Manage secret scope permissions](https://docs.databricks.com/aws/en/security/secrets/#manage-secret-scope-permissions) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Dash](https://pypi.org/project/dash/) - `dash`

```python title="requirements.txt"
databricks-sdk
dash
```

---

## OLTP Database


This recipe connects to a [Databricks Lakebase](https://docs.databricks.com/aws/en/oltp/) OLTP database instance to read data from PostgreSQL tables. It uses OAuth token-based authentication with connection pooling for efficient database access.

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

from psycopg_pool import ConnectionPool

w = WorkspaceClient()

class RotatingTokenConnection(psycopg.Connection):
    """psycopg3 Connection that injects a fresh OAuth token as the password."""
    
    @classmethod
    def connect(cls, conninfo: str = "", **kwargs):
        kwargs["password"] = w.database.generate_database_credential(
            request_id=str(uuid.uuid4()),
            instance_names=[kwargs.pop("_instance_name")]
        ).token
        kwargs.setdefault("sslmode", "require")
        return super().connect(conninfo, **kwargs)

def build_pool(instance_name: str, host: str, user: str, database: str) -> ConnectionPool:
    return ConnectionPool(
        conninfo=f"host={host} dbname={database} user={user}",
        connection_class=RotatingTokenConnection,
        kwargs={"_instance_name": instance_name},
        min_size=1,
        max_size=5,
        open=True,
    )

def query_df(pool: ConnectionPool, sql: str) -> pd.DataFrame:
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute(sql)
            if cur.description is None:
                return pd.DataFrame()
            cols = [d.name for d in cur.description]
            rows = cur.fetchall()
    return pd.DataFrame(rows, columns=cols)

# Usage
instance_name = "dbase_instance"
database = "databricks_postgres"
schema = "public"
table = "app_state"
user = w.current_user.me().user_name
host = w.database.get_database_instance(name=instance_name).read_write_dns

pool = build_pool(instance_name, host, user, database)

# Query existing data
df = query_df(pool, f"SELECT * FROM {schema}.{table} LIMIT 10")
```

## Requirements

### Permissions (app service principal)
- The database instance should be specified in your [App resources](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/resources).
- A PostgreSQL role for the service principal is **required**. See [this guide](https://docs.databricks.com/aws/en/oltp/pg-roles?language=PostgreSQL#create-postgres-roles-and-grant-privileges-for-databricks-identities).
- The PostgreSQL service principal role should have these example grants:

```sql
GRANT CONNECT ON DATABASE databricks_postgres TO "099f0306-9e29-4a87-84c0-3046e4bcea02";
GRANT USAGE, CREATE ON SCHEMA public TO "099f0306-9e29-4a87-84c0-3046e4bcea02";
GRANT SELECT ON TABLE app_state TO "099f0306-9e29-4a87-84c0-3046e4bcea02";
```

### Databricks resources
- [Lakebase](https://docs.databricks.com/aws/en/oltp/) database instance (PostgreSQL).
- Target PostgreSQL database/schema/table.

### Dependencies
- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk>=0.60.0`
- [`psycopg[binary]`](https://pypi.org/project/psycopg/), [`psycopg-pool`](https://pypi.org/project/psycopg-pool/)
- [Pandas](https://pypi.org/project/pandas/) - `pandas`
- [Dash](https://pypi.org/project/dash/) - `dash`

:::info
Tokens expire periodically; this app refreshes on each new connection and enforces TLS (sslmode=require).
:::

---

## Edit a Delta table


Use this recipe to read, edit, and write back data in a [Unity Catalog table](https://docs.databricks.com/aws/en/tables/) using the [Databricks SQL Connector](https://docs.databricks.com/en/dev-tools/python-sql-connector.html).

## Code snippet

```python title="app.py"
from functools import lru_cache
from databricks import sql
from databricks.sdk.core import Config

cfg = Config()

@lru_cache(maxsize=1)
def get_connection(http_path):
    return sql.connect(
        server_hostname=cfg.host,
        http_path=http_path,
        credentials_provider=lambda: cfg.authenticate,
    )

def read_table(table_name: str, conn) -> pd.DataFrame:
    with conn.cursor() as cursor:
        cursor.execute(f"SELECT * FROM {table_name}")
        return cursor.fetchall_arrow().to_pandas()

def insert_overwrite_table(table_name: str, df: pd.DataFrame, conn):
    with conn.cursor() as cursor:
        rows = list(df.itertuples(index=False, name=None))
        if not rows:
            return

        cols = list(df.columns)
        num_cols = len(cols)
        params = {}
        values_sql_parts = []
        p = 0
        for row in rows:
            ph = []
            for v in row:
                key = f"p{p}"
                ph.append(f":{key}")
                params[key] = v
                p += 1
            values_sql_parts.append("(" + ",".join(ph) + ")")

        values_sql = ",".join(values_sql_parts)
        col_list_sql = ",".join(cols)
        cursor.execute(f"INSERT OVERWRITE {table_name} ({col_list_sql}) VALUES {values_sql}", params)

http_path_input = "/sql/1.0/warehouses/xxxxxx"
table_name = "catalog.schema.table"
conn = get_connection(http_path_input)
df = read_table(table_name, conn)
# Edit the dataframe
insert_overwrite_table(table_name, df, conn)
```

:::info

This sample uses Pythons's [lru_cache](https://docs.python.org/3/library/functools.html#functools.lru_cache). Adapt the caching behavior to fit your specific use case.

:::

## Resources

- [SQL warehouse](https://docs.databricks.com/aws/en/compute/sql-warehouse/)
- [Unity Catalog table](https://docs.databricks.com/aws/en/tables/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `MODIFY` on the Unity Catalog table
- `CAN USE` on the SQL warehouse

See Unity [Catalog privileges and securable objects](https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges) for more information.

## Dependencies

- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Databricks SQL Connector](https://pypi.org/project/databricks-sql-connector/) - `databricks-sql-connector`
- [Pandas](https://pypi.org/project/pandas/) - `pandas`
- [Dash](https://pypi.org/project/dash/) - `dash`

```python title="requirements.txt"
databricks-sdk
databricks-sql-connector
pandas
dash
```

---

## Read a Delta table


This recipe reads a [Unity Catalog table](https://docs.databricks.com/aws/en/tables/) using the [Databricks SQL Connector](https://docs.databricks.com/en/dev-tools/python-sql-connector.html).

## Code snippet

```python title="app.py"
from functools import lru_cache
from databricks import sql
from databricks.sdk.core import Config

cfg = Config()  # Set the DATABRICKS_HOST environment variable when running locally

@lru_cache(maxsize=1)
def get_connection(http_path):
    return sql.connect(
        server_hostname=cfg.host,
        http_path=http_path,
        credentials_provider=lambda: cfg.authenticate,
    )

def read_table(table_name, conn):
    with conn.cursor() as cursor:
        query = f"SELECT * FROM {table_name}"
        cursor.execute(query)
        return cursor.fetchall_arrow().to_pandas()

http_path_input = "/sql/1.0/warehouses/xxxxxx"
table_name = "catalog.schema.table"
conn = get_connection(http_path_input)
df = read_table(table_name, conn)
```

:::info

This sample uses Pythons's [lru_cache](https://docs.python.org/3/library/functools.html#functools.lru_cache). Adapt the caching behavior to fit your specific use case.

:::

## Resources

- [SQL warehouse](https://docs.databricks.com/aws/en/compute/sql-warehouse/)
- [Unity Catalog table](https://docs.databricks.com/aws/en/tables/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `SELECT` on the Unity Catalog table
- `CAN USE` on the SQL warehouse

See Unity [Catalog privileges and securable objects](https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges) for more information.

## Dependencies

- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Databricks SQL Connector](https://pypi.org/project/databricks-sql-connector/) - `databricks-sql-connector`
- [Dash](https://pypi.org/project/dash/) - `dash`

```python title="requirements.txt"
databricks-sdk
databricks-sql-connector
dash
```

---

## Download a file


This recipe downloads a file from a [Unity Catalog volume](https://docs.databricks.com/en/volumes/index.html) using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/).

:::note

Unlike notebooks, Databricks Apps does not support mounting Unity Catalog volumes and directly reading and writing files. As this code snippet demonstrates, each file needs to be downloaded to the app compute before being able to manipulate it.

:::

## Code snippet

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

download_file_path = "/Volumes/catalog/schema/volume_name/file.csv"
response = w.files.download(download_file_path)
file_data = response.contents.read()
file_name = os.path.basename(download_file_path)
```

## Resources

- [Unity Catalog volume](https://docs.databricks.com/aws/en/files/volumes)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CATALOG` on the volume's catalog
- `USE SCHEMA` on the volume's schema
- `READ VOLUME` on the volume

See [Privileges required for volume operations](https://docs.databricks.com/en/volumes/privileges.html#privileges-required-for-volume-operations) for more information.

If you declare volume access in a Databricks Asset Bundle, `resources.apps[*].resources[*].uc_securable` may not grant `USE_CATALOG` and `USE_SCHEMA` on the parent catalog and schema (the app still needs them at runtime). As a temporary workaround until bundles can declare those parent grants, add the privileges manually, or see [apps_grants_sync](https://github.com/salihbout/apps_grants_sync): an example Databricks App and Asset Bundle that wires `experimental.scripts.postdeploy` so parent privileges are applied after each `databricks bundle deploy` (copy its `tools/` into your bundle or mirror the same pattern in `databricks.yml`).

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Dash](https://pypi.org/project/dash/) - `dash`

```python title="requirements.txt"
databricks-sdk
dash
```

---

## Upload a file


This recipe uploads a file to a [Unity Catalog volume](https://docs.databricks.com/en/volumes/index.html) using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/).

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# Read file into bytes
with open("local_file.csv", "rb") as f:
    file_bytes = f.read()
binary_data = io.BytesIO(file_bytes)

# Specify volume path and upload
volume_path = "main.marketing.raw_files"

parts = volume_path.strip().split(".")

catalog = parts[0]
schema = parts[1]
volume_name = parts[2]

volume_file_path = f"/Volumes/{catalog}/{schema}/{volume_name}/local_file.csv"
w.files.upload(volume_file_path, binary_data, overwrite=True)
```

## Resources

- [Unity Catalog volume](https://docs.databricks.com/aws/en/files/volumes)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CATALOG` on the catalog of the volume
- `USE SCHEMA` on the schema of the volume
- `READ VOLUME` and `WRITE VOLUME` on the volume

See [Privileges required for volume operations](https://docs.databricks.com/en/volumes/privileges.html#privileges-required-for-volume-operations) for more information.

If you declare volume access in a Databricks Asset Bundle, `resources.apps[*].resources[*].uc_securable` may not grant `USE_CATALOG` and `USE_SCHEMA` on the parent catalog and schema (the app still needs them at runtime). As a temporary workaround until bundles can declare those parent grants, add the privileges manually, or see [apps_grants_sync](https://github.com/salihbout/apps_grants_sync): an example Databricks App and Asset Bundle that wires `experimental.scripts.postdeploy` so parent privileges are applied after each `databricks bundle deploy` (copy its `tools/` into your bundle or mirror the same pattern in `databricks.yml`).

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Dash](https://pypi.org/project/dash/) - `dash`

```python title="requirements.txt"
databricks-sdk
dash
```

---

## Retrieve workflow results


This recipe retreives the results of a [Databricks Workflows](https://docs.databricks.com/en/jobs/index.html) job task run using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/)..

## Code snippet

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

task_run_id = "293894477334278"
results = w.jobs.get_run_output(task_run_id)

print(results)
```

## Resources

- [Job](https://docs.databricks.com/aws/en/jobs/configure-job)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN VIEW` permission on the job

See [Control access to a job](https://docs.databricks.com/en/jobs/privileges.html#control-access-to-a-job) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Dash](https://pypi.org/project/dash/) - `dash`

```python title="requirements.txt"
databricks-sdk
dash
```

---

## Run a workflow


This recipe triggers a [Databricks Workflows](https://docs.databricks.com/en/jobs/index.html) job using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/).

## Code snippet

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

job_id = "921773893211960"
parameters = {"param1": "value1", "param2": "value2"}

try:
    run = w.jobs.run_now(job_id=job_id, job_parameters=parameters)
    print(f"Started run with ID {run.run_id}")
except Exception as e:
    print(f"Error: {e}")
```

## Resources

- [Job](https://docs.databricks.com/aws/en/jobs/configure-job)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN MANAGE RUN` permission on the job

See [Control access to a job](https://docs.databricks.com/en/jobs/privileges.html#control-access-to-a-job) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Dash](https://pypi.org/project/dash/) - `dash`

```python title="requirements.txt"
databricks-sdk
dash
```

---

## Deployment instructions


Follow these instructions to deploy the interactive examples to your Databricks workspace or to run them locally.

:::warning

These samples are experimental and meant for demonstration purposes only. They are provided as-is and without formal support by Databricks. Ensure your organization's security, compliance, and operational best practices are applied before deploying them to production.

:::

## Deploy to Databricks

1. Navigate to the [databricks-apps-cookbook](https://github.com/databricks-solutions/databricks-apps-cookbook) GitHub repository and [load it a Databricks Git folder](https://docs.databricks.com/en/repos/index.html) in your Databricks workspace.
1. In your Databricks workspace, switch to **Compute** -> **Apps**.
1. Choose **Create app**.
1. Under **Choose how to start**, select **Custom** and choose **Next**.
1. Provide a name for your app and choose **Create app**.
1. Once your app compute has started, choose **Deploy**.
1. Navigate to your new Git folder and select either the `dash` or `streamlit` folder.
1. Choose **Deploy**.

:::info

Check the Requirements tab of each recipe to understand what [service principal permissions](https://docs.databricks.com/en/dev-tools/databricks-apps/app-development.html#configure-resources), Databricks resources, and Python packages are required to use it.

:::

## Run locally

1. [Clone](https://docs.github.com/en/repositories/creating-and-managing-repositories/cloning-a-repository) the [databricks-apps-cookbook](https://github.com/databricks-solutions/databricks-apps-cookbook) GitHub repository or your fork to your local machine and switch into the `databricks-apps-cookbook` folder:
   ```bash
   git clone https://github.com/databricks-solutions/databricks-apps-cookbook.git
   cd databricks-apps-cookbook
   ```
1. Navigate to the sub-folder for the cookbook framework you want to run (either `dash` or `streamlit` or `reflex`). Create and activate a Python virtual environment in this folder [`venv`](https://docs.python.org/3/library/venv.html). We recommend using separate environments for each framework:
   ```bash
   cd reflex
   python3 -m venv .venv
   source .venv/bin/activate
   ```
1. Install required packages:
   ```bash
   pip install -r requirements.txt
   ```
1. Install the [Databricks CLI](https://docs.databricks.com/en/dev-tools/cli/index.html) and authenticate with your Databricks workspace using [OAuth U2M](https://docs.databricks.com/en/dev-tools/auth/oauth-u2m.html), for example:
   ```bash
   databricks auth login --host https://my-workspace.cloud.databricks.com/
   ```
1. Set required environment variables:
   ```bash
   export DATABRICKS_HOST=https://my-workspace.cloud.databricks.com/
   ```
1. Run the cookbook app locally (make sure your virtual environment is activated).

   Streamlit:

   ```bash
   streamlit run app.py
   ```
   
   Reflex:

   ```bash
   reflex run app.py
   ```

   Dash:

   ```bash
   python app.py
   ```

   FastAPI:

   ```bash
   uvicorn app:app
   ```

:::info

Make sure you have a working network connection to your Databricks workspace. Some samples may only work when running on Databricks Apps and not locally, e.g., retrieving information from HTTP headers to identify users.

:::

---

## Interact with Lakebase Tables


This recipe demonstrates how to build a complete orders management API using Lakebase PostgreSQL database. These endpoints provide CRUD operations and various query patterns for handling orders data synchronized from Databricks Unity Catalog.

:::info Prerequisites
- Lakebase resources must be created using the [Create Lakebase Resources](./lakebase_resources_create.mdx) endpoint
- The synced table pipeline must be completed and orders data synchronized from `samples.tpch.orders`
- Database connection must be configured in your application environment
:::

:::info
In this example, we demonstrate multiple HTTP methods (GET, POST) which are the standard choices for data operations in REST APIs as defined in RFC 7231:
- `GET` requests are idempotent and cacheable, ideal for data retrieval
- `POST` requests for updates and modifications

For detailed specifications, refer to [RFC 7231](https://datatracker.ietf.org/doc/html/rfc7231) which defines HTTP method semantics.
:::

## Code snippet

```python title="routes/v1/orders.py"

from config.database import get_async_db
from models.orders import (
    CursorPaginationInfo,
    Order,
    OrderCount,
    OrderListCursorResponse,
    OrderListResponse,
    OrderRead,
    OrderSample,
    OrderStatusUpdate,
    OrderStatusUpdateResponse,
    PaginationInfo,
)
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from fastapi import APIRouter, Depends, HTTPException, Query

logger = logging.getLogger(__name__)
router = APIRouter(tags=["orders"])

# 1. GET ORDERS COUNT
@router.get("/count", response_model=OrderCount, summary="Get total order count")
async def get_order_count(db: AsyncSession = Depends(get_async_db)):
    try:
        stmt = select(func.count(Order.o_orderkey))
        result = await db.execute(stmt)
        count = result.scalar()
        return OrderCount(total_orders=count)
    except Exception as e:
        logger.error(f"Error getting order count: {e}")
        raise HTTPException(status_code=500, detail="Failed to retrieve order count")

# 2. GET SAMPLE ORDERS
@router.get("/sample", response_model=OrderSample, summary="Get 5 random order keys")
async def get_sample_orders(db: AsyncSession = Depends(get_async_db)):
    try:
        stmt = select(Order.o_orderkey).limit(5)
        result = await db.execute(stmt)
        order_keys = result.scalars().all()
        return OrderSample(sample_order_keys=order_keys)
    except Exception as e:
        logger.error(f"Error getting sample orders: {e}")
        raise HTTPException(status_code=500, detail="Failed to retrieve sample orders")

# 3. PAGE-BASED PAGINATION
@router.get("/pages", response_model=OrderListResponse, summary="Get orders with page-based pagination")
async def get_orders_by_page(
    page: int = Query(1, ge=1, description="Page number (1-based)"),
    page_size: int = Query(100, ge=1, le=1000, description="Number of records per page (max 1000)"),
    include_count: bool = Query(True, description="Include total count for pagination info"),
    db: AsyncSession = Depends(get_async_db),
):
    try:
        if include_count:
            count_stmt = select(func.count(Order.o_orderkey))
            count_result = await db.execute(count_stmt)
            total_count = count_result.scalar()
            total_pages = (total_count + page_size - 1) // page_size
        else:
            total_count = -1
            total_pages = -1

        offset = (page - 1) * page_size
        stmt = (
            select(Order)
            .order_by(Order.o_orderkey)
            .offset(offset)
            .limit(page_size + 1)
        )

        result = await db.execute(stmt)
        all_orders = result.scalars().all()

        has_next = len(all_orders) > page_size
        orders = all_orders[:page_size]
        has_previous = page > 1

        pagination_info = PaginationInfo(
            page=page,
            page_size=page_size,
            total_pages=total_pages,
            total_count=total_count,
            has_next=has_next,
            has_previous=has_previous,
        )

        return OrderListResponse(orders=orders, pagination=pagination_info)

    except Exception as e:
        logger.error(f"Error getting page-based orders: {e}")
        raise HTTPException(status_code=500, detail="Failed to retrieve orders")

# 4. CURSOR-BASED PAGINATION
@router.get("/stream", response_model=OrderListCursorResponse, summary="Get orders with cursor-based pagination")
async def get_orders_by_cursor(
    cursor: int = Query(0, ge=0, description="Start after this order key (0 for beginning)"),
    page_size: int = Query(100, ge=1, le=1000, description="Number of records to fetch (max 1000)"),
    db: AsyncSession = Depends(get_async_db),
):
    try:
        stmt = (
            select(Order)
            .where(Order.o_orderkey > cursor)
            .order_by(Order.o_orderkey)
            .limit(page_size + 1)
        )

        result = await db.execute(stmt)
        all_orders = result.scalars().all()

        has_next = len(all_orders) > page_size
        orders = all_orders[:page_size]
        has_previous = cursor > 0

        next_cursor = orders[-1].o_orderkey if orders and has_next else None
        previous_cursor = max(0, cursor - page_size) if has_previous else None

        pagination_info = CursorPaginationInfo(
            page_size=page_size,
            has_next=has_next,
            has_previous=has_previous,
            next_cursor=next_cursor,
            previous_cursor=previous_cursor,
        )

        return OrderListCursorResponse(orders=orders, pagination=pagination_info)

    except Exception as e:
        logger.error(f"Error getting cursor-based orders: {e}")
        raise HTTPException(status_code=500, detail="Failed to retrieve orders")

# 5. GET SPECIFIC ORDER
@router.get("/{order_key}", response_model=OrderRead, summary="Get an order by its key")
async def read_order(order_key: int, db: AsyncSession = Depends(get_async_db)):
    try:
        if order_key <= 0:
            raise HTTPException(status_code=400, detail="Invalid order key provided")

        stmt = select(Order).where(Order.o_orderkey == order_key)
        result = await db.execute(stmt)
        order = result.scalars().first()

        if not order:
            raise HTTPException(status_code=404, detail=f"Order with key '{order_key}' not found")

        return order

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Unexpected error fetching order {order_key}: {e}")
        raise HTTPException(status_code=500, detail="Internal server error occurred")

# 6. UPDATE ORDER STATUS
@router.post("/{order_key}/status", response_model=OrderStatusUpdateResponse, summary="Update order status")
async def update_order_status(
    order_key: int,
    status_data: OrderStatusUpdate,
    db: AsyncSession = Depends(get_async_db),
):
    try:
        if order_key <= 0:
            raise HTTPException(status_code=400, detail="Invalid order key provided")

        check_stmt = select(Order).where(Order.o_orderkey == order_key)
        check_result = await db.execute(check_stmt)
        existing_order = check_result.scalars().first()

        if not existing_order:
            raise HTTPException(status_code=404, detail=f"Order with key '{order_key}' not found")

        existing_order.o_orderstatus = status_data.o_orderstatus
        await db.commit()
        await db.refresh(existing_order)

        return OrderStatusUpdateResponse(
            o_orderkey=order_key,
            o_orderstatus=status_data.o_orderstatus,
            message="Order status updated successfully",
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error updating status for order {order_key}: {e}")
        raise HTTPException(status_code=500, detail="Failed to update order status")
```

:::warning

The above example is shortened for brevity and not suitable for production use.
You can find a more advanced sample in the databricks-apps-cookbook GitHub repository.

:::

## Example Usage

The orders API provides six main endpoints for different use cases:

| Endpoint | Method | Purpose | Best For |
|----------|--------|---------|----------|
| `/orders/count` | GET | Get total orders count | Dashboards, monitoring |
| `/orders/sample` | GET | Get 5 sample order keys | Testing, development |
| `/orders/pages` | GET | Page-based pagination | Traditional UIs with page numbers |
| `/orders/stream` | GET | Cursor-based pagination | Large datasets, infinite scroll |
| `/orders/{order_key}` | GET | Get specific order | Order details, lookups |
| `/orders/{order_key}/status` | POST | Update order status | Order processing workflows |

### Get Orders Count

```bash
curl -X GET "http://localhost:8000/api/v1/orders/count"
```

```json
{
  "total_orders": 1500000
}
```

### Get Sample Orders

```bash
curl -X GET "http://localhost:8000/api/v1/orders/sample"
```

```json
{
  "sample_order_keys": [1, 32, 33, 34, 35]
}
```

### Page-Based Pagination

```bash
curl -X GET "http://localhost:8000/api/v1/orders/pages?page=1&page_size=2"
```

```json
{
  "orders": [
    {
      "o_orderkey": 1,
      "o_custkey": 370,
      "o_orderstatus": "O",
      "o_totalprice": 172799.49,
      "o_orderdate": "1996-01-02",
      "o_orderpriority": "5-LOW",
      "o_clerk": "Clerk#000000951",
      "o_shippriority": 0,
      "o_comment": "nstructions sleep furiously among"
    },
    {
      "o_orderkey": 2,
      "o_custkey": 781,
      "o_orderstatus": "O",
      "o_totalprice": 46929.18,
      "o_orderdate": "1996-12-01",
      "o_orderpriority": "1-URGENT",
      "o_clerk": "Clerk#000000880",
      "o_shippriority": 0,
      "o_comment": "foxes. pending accounts at the pending"
    }
  ],
  "pagination": {
    "page": 1,
    "page_size": 2,
    "total_pages": 750000,
    "total_count": 1500000,
    "has_next": true,
    "has_previous": false
  }
}
```

### Update Order Status

```bash
curl -X POST "http://localhost:8000/api/v1/orders/1/status" \
  -H "Content-Type: application/json" \
  -d '{"o_orderstatus": "F"}'
```

```json
{
  "o_orderkey": 1,
  "o_orderstatus": "F",
  "message": "Order status updated successfully"
}
```

## Resources

- [Lakebase PostgreSQL](https://docs.databricks.com/en/database/index.html)
- [SQLAlchemy Async](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html)
- [FastAPI Dependencies](https://fastapi.tiangolo.com/tutorial/dependencies/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- Database instance connection access via OAuth tokens
- `SELECT` permissions on the `orders_synced` table in your Lakebase database
- `UPDATE` permissions on the `orders_synced` table for status updates
- Database user role with appropriate table access

See [Lakebase permissions](https://docs.databricks.com/en/database/permissions.html) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [FastAPI](https://pypi.org/project/fastapi/) - `fastapi`
- [SQLAlchemy](https://pypi.org/project/sqlalchemy/) - `sqlalchemy`
- [SQLModel](https://pypi.org/project/sqlmodel/) - `sqlmodel`
- [asyncpg](https://pypi.org/project/asyncpg/) - `asyncpg`
- [uvicorn](https://pypi.org/project/uvicorn/) - `uvicorn`

```python title="requirements.txt"
databricks-sdk>=0.60.0
fastapi
sqlalchemy
sqlmodel
asyncpg
uvicorn
```

---

## Create Lakebase Resources


This recipe demonstrates how to programmatically create Lakebase PostgreSQL resources in your Databricks workspace using FastAPI. This endpoint sets up a complete Lakebase environment including a database instance, catalog, and synced table pipeline.

:::warning Cost Alert
This endpoint creates billable resources in your Databricks environment including:
- A Lakebase PostgreSQL database instance
- A synced table pipeline

These resources will incur ongoing costs until deleted. Monitor your usage and delete resources when no longer needed using the delete endpoint.
:::

:::info
In this example, we set up our API to be called using the `POST` HTTP method which is the standard choice for creating new resources in REST APIs as defined in RFC 7231.
Unlike `GET`, POST is not idempotent - making the same request multiple times may create multiple resources.

For detailed specifications, refer to [RFC 7231 Section 4.3.3](https://datatracker.ietf.org/doc/html/rfc7231#section-4.3.3) which defines the POST method's semantics and requirements.
:::

## Code snippet

```python title="routes/v1/lakebase.py"

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.database import (
    DatabaseCatalog,
    DatabaseInstance,
    DatabaseInstanceRole,
    DatabaseInstanceRoleAttributes,
    DatabaseInstanceRoleIdentityType,
    DatabaseInstanceRoleMembershipRole,
    NewPipelineSpec,
    SyncedDatabaseTable,
    SyncedTableSchedulingPolicy,
    SyncedTableSpec,
)
from models.lakebase import LakebaseResourcesResponse

from fastapi import APIRouter, HTTPException, Query

# Environment variables used:
# LAKEBASE_INSTANCE_NAME=my-lakebase-instance
# LAKEBASE_DATABASE_NAME=my_database  
# LAKEBASE_CATALOG_NAME=my-pg-catalog
# SYNCHED_TABLE_STORAGE_CATALOG=my_storage_catalog
# SYNCHED_TABLE_STORAGE_SCHEMA=my_storage_schema

logger = logging.getLogger(__name__)
w = WorkspaceClient()
router = APIRouter(tags=["lakebase"])
current_user_id = w.current_user.me().id

@router.post(
    "/resources/create-lakebase-resources",
    response_model=LakebaseResourcesResponse,
    summary="Create Lakebase Resources",
)
async def create_lakebase_resources(
    create_resources: bool = Query(
        description="""🚨 This endpoint creates resources in your Databricks environment that will incur a cost. 
        By setting this value to true you understand the costs associated with this action. 🚨
        ⌛️ This endpoint may take a few minutes to complete.⌛️""",
    ),
    capacity: str = Query("CU_1", description="Capacity of the Lakebase instance"),
    node_count: int = Query(1, description="Number of nodes in the Lakebase instance"),
    enable_readable_secondaries: bool = Query(
        False, description="Enable readable secondaries"
    ),
    retention_window_in_days: int = Query(
        7, description="Retention window in days for the Lakebase instance"
    ),
):
    if not create_resources:
        return LakebaseResourcesResponse(
            instance="",
            catalog="",
            synced_table="",
            message="No resources were created (create_resources=False)",
        )

    instance_name = os.getenv("LAKEBASE_INSTANCE_NAME", f"{current_user_id}-lakebase-demo")

    # Check if instance already exists
    try:
        instance_exists = w.database.get_database_instance(name=instance_name)
        logger.info(f"Instance {instance_name} already exists. Skipping creation.")
        return LakebaseResourcesResponse(
            instance=instance_name,
            catalog="",
            synced_table="",
            message="Instance already exists, skipping creation.",
        )
    except Exception as e:
        if "not found" not in str(e).lower():
            raise HTTPException(status_code=500, detail=f"Error checking instance existence: {str(e)}")

    # Create database instance
    instance = DatabaseInstance(
        name=instance_name,
        capacity=capacity,
        node_count=node_count,
        enable_readable_secondaries=enable_readable_secondaries,
        retention_window_in_days=retention_window_in_days,
    )
    instance_create = w.database.create_database_instance_and_wait(instance)

    # Create superuser role
    superuser_role = DatabaseInstanceRole(
        name=w.current_user.me().user_name,
        identity_type=DatabaseInstanceRoleIdentityType.USER,
        membership_role=DatabaseInstanceRoleMembershipRole.DATABRICKS_SUPERUSER,
        attributes=DatabaseInstanceRoleAttributes(bypassrls=True, createdb=True, createrole=True),
    )

    try:
        w.database.create_database_instance_role(
            instance_name=instance_create.name,
            database_instance_role=superuser_role,
        )
    except Exception as e:
        logger.error(f"Failed to create superuser role: {e}")

    # Create catalog
    lakebase_database_name = os.getenv("LAKEBASE_DATABASE_NAME", "demo_database")
    catalog_name = os.getenv("LAKEBASE_CATALOG_NAME", f"{current_user_id}-pg-catalog")
    
    catalog = DatabaseCatalog(
        name=catalog_name,
        database_instance_name=instance_create.name,
        database_name=lakebase_database_name,
        create_database_if_not_exists=True,
    )
    database_create = w.database.create_database_catalog(catalog)

    # Create synced table
    synced_table_storage_catalog = os.getenv("SYNCHED_TABLE_STORAGE_CATALOG", "default_storage_catalog")
    synced_table_storage_schema = os.getenv("SYNCHED_TABLE_STORAGE_SCHEMA", "default_storage_schema")

    new_pipeline = NewPipelineSpec(
        storage_catalog=synced_table_storage_catalog,
        storage_schema=synced_table_storage_schema,
    )

    spec = SyncedTableSpec(
        source_table_full_name="samples.tpch.orders",
        primary_key_columns=["o_orderkey"],
        timeseries_key="o_orderdate",
        create_database_objects_if_missing=True,
        new_pipeline_spec=new_pipeline,
        scheduling_policy=SyncedTableSchedulingPolicy.SNAPSHOT,
    )

    synced_table = SyncedDatabaseTable(
        name=f"{catalog_name}.public.orders_synced",
        database_instance_name=instance_create.name,
        logical_database_name=lakebase_database_name,
        spec=spec,
    )

    try:
        synced_table_create = w.database.create_synced_database_table(synced_table)
        pipeline_id = synced_table_create.id
    except Exception as e:
        logger.error(f"API error during synced table creation: {e}")
        pipeline_id = "check-workspace-ui"

    workspace_url = w.config.host
    if pipeline_id != "check-workspace-ui":
        pipeline_url = f"{workspace_url}/pipelines/{pipeline_id}"
        message = f"Resources created successfully. Synced table pipeline {pipeline_id} is provisioning asynchronously. Monitor progress at: {pipeline_url}"
    else:
        message = f"Resources created successfully. Synced table pipeline initiated. Check pipelines in workspace: {workspace_url}/pipelines"

    return LakebaseResourcesResponse(
        instance=instance_create.name,
        catalog=database_create.name,
        synced_table=pipeline_id,
        message=message,
    )
```

:::warning

The above example is shortened for brevity and not suitable for production use.
You can find a more advanced sample in the databricks-apps-cookbook GitHub repository.

:::

## Example Usage

```bash
# Create Lakebase resources with default settings
curl -X POST "http://localhost:8000/api/v1/resources/create-lakebase-resources?create_resources=true"

# Create with custom capacity and multiple nodes
curl -X POST "http://localhost:8000/api/v1/resources/create-lakebase-resources?create_resources=true&capacity=CU_2&node_count=2&enable_readable_secondaries=true"
```

If the request was successful, you will get the following output:

```json
{
  "instance": "user123-lakebase-demo",
  "catalog": "user123-pg-catalog", 
  "synced_table": "pipeline-abc123",
  "message": "Resources created successfully. Synced table pipeline abc123 is provisioning asynchronously. Monitor progress at: https://workspace.databricks.com/pipelines/abc123"
}
```

## Resources

- [Lakebase PostgreSQL](https://docs.databricks.com/en/database/index.html)
- [Unity Catalog](https://docs.databricks.com/en/data-governance/unity-catalog/index.html)
- [Synced Tables](https://docs.databricks.com/en/database/synced-tables.html)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CREATE` permissions on the Unity Catalog metastore
- `CAN USE` on the storage catalog and schema for synced tables
- Access to the source table `samples.tpch.orders`

See [Lakebase permissions](https://docs.databricks.com/en/database/permissions.html) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [FastAPI](https://pypi.org/project/fastapi/) - `fastapi`
- [uvicorn](https://pypi.org/project/uvicorn/) - `uvicorn`

```python title="requirements.txt"
databricks-sdk>=0.60.0
fastapi
uvicorn
```

---

## Delete Lakebase Resources


This recipe demonstrates how to programmatically delete Lakebase PostgreSQL resources from your Databricks workspace using FastAPI. This endpoint safely removes all created Lakebase resources to avoid ongoing costs.

:::warning Destructive Operation
This endpoint permanently deletes:
- Database catalog and all its data
- PostgreSQL database instance and all stored data

**This action cannot be undone.** Ensure you have backed up any important data before deletion.

The pipeline created to sync orders to our postgres instance will need to be **deleted manually.**
:::

:::info
In this example, we set up our API to be called using the `DELETE` HTTP method which is the standard choice for removing resources in REST APIs as defined in RFC 7231.
DELETE requests are idempotent - making the same request multiple times has the same effect as making it once.

For detailed specifications, refer to [RFC 7231 Section 4.3.5](https://datatracker.ietf.org/doc/html/rfc7231#section-4.3.5) which defines the DELETE method's semantics and requirements.
:::

## Code snippet

```python title="routes/v1/lakebase.py"

from databricks.sdk import WorkspaceClient
from models.lakebase import LakebaseResourcesDeleteResponse

from fastapi import APIRouter, Query

# Environment variables used:
# LAKEBASE_INSTANCE_NAME=my-lakebase-instance
# LAKEBASE_CATALOG_NAME=my-pg-catalog

logger = logging.getLogger(__name__)
w = WorkspaceClient()
router = APIRouter(tags=["lakebase"])
current_user_id = w.current_user.me().id

@router.delete(
    "/resources/delete-lakebase-resources",
    response_model=LakebaseResourcesDeleteResponse,
    summary="Delete Lakebase Resources",
)
async def delete_lakebase_resources(
    confirm_deletion: bool = Query(
        description="""🚨 This endpoint will permanently delete Lakebase resources. 
        Set to true to confirm you want to delete these resources. 🚨
        ⌛️ This endpoint may take a few minutes to complete.⌛️""",
    ),
):
    if not confirm_deletion:
        return LakebaseResourcesDeleteResponse(
            deleted_resources=[],
            failed_deletions=[],
            message="No resources were deleted (confirm_deletion=False)",
        )

    instance_name = os.getenv("LAKEBASE_INSTANCE_NAME", f"{current_user_id}-lakebase-demo")
    catalog_name = os.getenv("LAKEBASE_CATALOG_NAME", f"{current_user_id}-pg-catalog")
    synced_table_name = f"{catalog_name}.public.orders_synced"

    deleted_resources = []
    failed_deletions = []

    # Delete synced table
    logger.info(f"Attempting to delete synced table: {synced_table_name}")
    try:
        w.database.delete_synced_database_table(name=synced_table_name)
        deleted_resources.append(f"Synced table: {synced_table_name}")
        logger.info(f"Successfully deleted synced table: {synced_table_name}")
    except Exception as e:
        failed_deletions.append(f"Synced table: {synced_table_name} - {str(e)}")
        logger.error(f"Failed to delete synced table {synced_table_name}: {e}")

    # Delete catalog
    logger.info(f"Attempting to delete catalog: {catalog_name}")
    try:
        w.database.delete_database_catalog(name=catalog_name)
        deleted_resources.append(f"Catalog: {catalog_name}")
        logger.info(f"Successfully deleted catalog: {catalog_name}")
    except Exception as e:
        failed_deletions.append(f"Catalog: {catalog_name} - {str(e)}")
        logger.error(f"Failed to delete catalog {catalog_name}: {e}")

    # Delete database instance
    logger.info(f"Attempting to delete database instance: {instance_name}")
    try:
        w.database.delete_database_instance(name=instance_name, purge=True)
        deleted_resources.append(f"Database instance: {instance_name}")
        logger.info(f"Successfully deleted database instance: {instance_name}")
    except Exception as e:
        failed_deletions.append(f"Database instance: {instance_name} - {str(e)}")
        logger.error(f"Failed to delete database instance {instance_name}: {e}")

    if failed_deletions:
        message = f"Deletion completed with errors. {len(deleted_resources)} resources deleted, {len(failed_deletions)} failed."
    else:
        message = f"All {len(deleted_resources)} resources deleted successfully."

    return LakebaseResourcesDeleteResponse(
        deleted_resources=deleted_resources,
        failed_deletions=failed_deletions,
        message=message,
    )
```

:::warning

The above example is shortened for brevity and not suitable for production use.
You can find a more advanced sample in the databricks-apps-cookbook GitHub repository.

:::

## Example Usage

```bash
# Delete all Lakebase resources
curl -X DELETE "http://localhost:8000/api/v1/resources/delete-lakebase-resources?confirm_deletion=true"

# Safe call without deletion (for testing)
curl -X DELETE "http://localhost:8000/api/v1/resources/delete-lakebase-resources?confirm_deletion=false"
```

If the request was successful, you will get the following output:

```json
{
  "deleted_resources": [
    "Synced table: my-pg-catalog.public.orders_synced",
    "Catalog: my-pg-catalog",
    "Database instance: my-lakebase-instance"
  ],
  "failed_deletions": [],
  "message": "All 3 resources deleted successfully."
}
```

## Resources

- [Lakebase PostgreSQL](https://docs.databricks.com/en/database/index.html)
- [Unity Catalog](https://docs.databricks.com/en/data-governance/unity-catalog/index.html)
- [Synced Tables](https://docs.databricks.com/en/database/synced-tables.html)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- Workspace admin privileges to delete database instances
- `DROP` permissions on the Unity Catalog objects
- Ownership or admin access to the database instance
- Access to delete synced table pipelines

See [Lakebase permissions](https://docs.databricks.com/en/database/permissions.html) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [FastAPI](https://pypi.org/project/fastapi/) - `fastapi`
- [uvicorn](https://pypi.org/project/uvicorn/) - `uvicorn`

```python title="requirements.txt"
databricks-sdk>=0.60.0
fastapi
uvicorn
```

---

## Connect an MCP server(Building_endpoints)


This recipe connects to an [MCP](https://modelcontextprotocol.io/overview) server for AI applications using GitHub as an example and the Unity Catalog [HTTP connection](https://docs.databricks.com/aws/en/query-federation/http) for secure and governed access.

## Code snippets

### OAuth User to Machine Per User (On-behalf-of-user)

```python

from pathlib import Path
from mcp.server.fastmcp import FastMCP
from fastapi import FastAPI, Request
from fastapi.responses import FileResponse
from typing import Dict, Any
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ExternalFunctionRequestHttpMethod

STATIC_DIR = Path(__file__).parent / "static"

# Create an MCP server
mcp = FastMCP("Custom MCP Server on Databricks Apps")

# Context variable for headers
header_store = contextvars.ContextVar("header_store")

def init_github_tools() -> Dict[str, Any]:
    headers = header_store.get({})
    token=headers["x-forwarded-access-token"]

    json = {
        "jsonrpc": "2.0",
        "id": "init-1",
        "method": "initialize",
        "params": {}
    }

    try:
        response =  WorkspaceClient(token=token, auth_type="pat").serving_endpoints.http_request(
            conn="github_u2m_connection",
            method=ExternalFunctionRequestHttpMethod.POST,
            path="/",
            json=json,
        )
    except Exception as e:
        print(f"Error making the init request: {e}")
        return {"error": str(e)}

    # Extract the Mcp-Session-Id from response headers
    print("MCP_HEADERS", response.headers)
    session_id = response.headers.get("mcp-session-id")
    if not session_id:
        print("No session ID returned by server.")

    print(f"✅ Got MCP Session ID: {session_id}")
    return session_id

# Tool using header_store
def list_github_tools() -> Dict[str, Any]:
    session_id = init_github_tools()
    headers = header_store.get({})
    token=headers["x-forwarded-access-token"]

    json = {
        "jsonrpc": "2.0",
        "id": "list-1",
        "method": "tools/list",
    }

    try:
        response =  WorkspaceClient(token=token, auth_type="pat").serving_endpoints.http_request(
            conn="github_u2m_connection",
            method=ExternalFunctionRequestHttpMethod.POST,
            path="/",
            json=json,
            headers={
                "Mcp-Session-Id": session_id
            }
        )
    except Exception as e:
        print(f"Error: {e}")
        return {"error": str(e)}

    print(f"Response list tools: {response.json()}")

    return response.json()

def call_github_tool(name: str, arguments: dict) -> Dict[str, Any]:
    session_id = init_github_tools()
    headers = header_store.get({})
    token=headers["x-forwarded-access-token"]
    json = {
        "jsonrpc": "2.0",
        "id": "call-1",
        "method": "tools/call",
        "params": {
            "name": name,
            "arguments": arguments
        }
    }
    try:
        response =  WorkspaceClient(token=token, auth_type="pat").serving_endpoints.http_request(
            conn="github_u2m_connection",
            method=ExternalFunctionRequestHttpMethod.POST,
            path="/",
            json=json,
            headers={
                "Mcp-Session-Id": session_id
            }
        )
    except Exception as e:
        print(f"Error: {e}")
        return {"error": str(e)}
    
    print(f"Response call tools: {response.json()}")
    return response.json()

@mcp._mcp_server.list_tools()
async def list_tools():
    tools_dict = list_github_tools()
    return tools_dict.get("result", {}).get("tools", [])

@mcp._mcp_server.call_tool()
async def call_tool(name: str, arguments: dict):
    response = call_github_tool(name, arguments)
    return response.get("result", {}).get("content", [])

mcp_app = mcp.streamable_http_app()

app = FastAPI(
    lifespan=lambda _: mcp.session_manager.run(),
)

@app.middleware("http")
async def capture_headers(request: Request, call_next):
    header_store.set(dict(request.headers))
    return await call_next(request)

@app.get("/", include_in_schema=False)
async def serve_index():
    return FileResponse(STATIC_DIR / "index.html")

app.mount("/", mcp_app)
```

## Resources

- [Unity Catalog HTTP Connection](https://docs.databricks.com/aws/en/query-federation/http) with the MCP (/mcp) base path

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CONNECTION` permission on the HTTP Connection

When using OAuth User to Machine Per User (On-behalf-of-user), you need to configure [User authorization](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/auth#user-authorization) by adding the Unity Catalog connection or other scopes.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [FastAPI](https://pypi.org/project/fastapi/) - `fastapi`
- [uvicorn](https://pypi.org/project/uvicorn/) - `uvicorn`
- [MCP CLI](https://pypi.org/project/mcp/) - `mcp[cli]`

```python title="requirements.txt"
databricks-sdk
fastapi
uvicorn
mcp[cli]
```

---

## Insert data into a table


This recipe demonstrates how to insert data into a Databricks [Unity Catalog table](https://docs.databricks.com/aws/en/tables/) from a FastAPI application using the [Databricks SQL Connector](https://docs.databricks.com/en/dev-tools/python-sql-connector.html).

:::info
In this example, we set up our API to be called using the `POST` HTTP method which is the standard choice for creating new resources in REST APIs as defined in RFC 7231.
Unlike `GET`, POST is not idempotent - making the same request multiple times may create multiple resources.

POST requests are typically used when submitting data to be processed or when creating new resources, with the request body containing the data to be added.

For detailed specifications, refer to [RFC 7231 Section 4.3.3](https://datatracker.ietf.org/doc/html/rfc7231#section-4.3.3) which defines the POST method's semantics and requirements.

:::

## Code snippet

```python title="app.py"

from typing import Dict, List

from fastapi import FastAPI, Request

from databricks import sql
from databricks.sdk.core import Config

DATABRICKS_WAREHOUSE_ID = os.environ.get("DATABRICKS_WAREHOUSE_ID") or None

app = FastAPI()
databricks_cfg = Config()

def get_connection(warehouse_id: str):
    http_path = f"/sql/1.0/warehouses/{DATABRICKS_WAREHOUSE_ID}"
    return sql.connect(
        server_hostname=databricks_cfg.host,
        http_path=http_path,
        credentials_provider=lambda: databricks_cfg.authenticate,
    )

def insert_data(table_path: str, data: List[Dict], warehouse_id: str) -> int:
    conn = get_connection(warehouse_id)
    try:
        with conn.cursor() as cursor:
            # Get columns
            columns = list(data[0].keys())
            columns_str = ", ".join(columns)
            placeholders = ", ".join(["?"] * len(columns))

            # Build the INSERT statement with multiple VALUES clauses
            values_clauses = []
            all_values = []

            for record in data:
                values_clauses.append(f"({placeholders})")
                all_values.extend(record[col] for col in columns)

            insert_query = f"""
                INSERT INTO {table_path} ({columns_str})
                VALUES {", ".join(values_clauses)}
            """

            # Execute the insert with all values in a single statement
            print(f"Executing query: {insert_query}")
            cursor.execute(insert_query, all_values)
            return cursor.rowcount

    except Exception as e:
        raise Exception(f"Failed to insert data: {str(e)}")

@app.post("/api/v1/table")
async def insert_table_data(request: Request):
    request_as_json = await request.json()
    request_as_dict = dict(request_as_json)
    results = None
    try:
        # Build the table path
        table_path = f"{request_as_dict['catalog']}.{request_as_dict['schema']}.{request_as_dict['table']}"

        # Insert the data
        records_inserted = insert_data(
            table_path=table_path,
            data=request_as_dict["data"],
            warehouse_id=DATABRICKS_WAREHOUSE_ID,
        )

        # Ensure records_inserted is not negative (DBSQL Side Effect)
        if records_inserted < 0:
            records_inserted = len(request_as_dict["data"])

        # Create the response
        results = {"data": request_as_dict["data"], "count": records_inserted}
    except Exception as e:
        raise Exception(f"FastAPI Request Failed: {str(e)}")

    return {"results": results}
```

:::warning

The above example is shortened for brevity and not suitable for production use.
You can find a more advanced sample in the databricks-apps-cookbook GitHub repository.

:::

### Example Usage

In this example, we will create the following Unity Catalog table called `my_catalog.my_schema.trips` via Databricks SQL Editor.  
It is assumed that the user/service principal identity has the appropriate Unity Catalog grants to make changes as required.

```sql
CREATE OR REPLACE TABLE my_catalog.my_schema.trips (
  trip_id INT,
  passenger_count INT,
  trip_distance FLOAT,
  pickup_datetime TIMESTAMP,
  dropoff_datetime TIMESTAMP,
  payment_type STRING,
  fare_amount FLOAT,
  tip_amount FLOAT
)
```

Once the table has been created, you can provide data (list of dicts) to be inserted using the API example provided above.  
To highlight this, please consult the example Python script below, noting the `POST` verb and `JSON` data payload.

```python title="insert_data_into_table.py"
from databricks.sdk.core import Config

config = Config(profile="my-env")
token = config.oauth_token().access_token

rows_to_be_inserted = [
    {
        "trip_id": 1,
        "passenger_count": 1,
        "trip_distance": 10.0,
        "pickup_datetime": "2024-01-01 12:00:00",
        "dropoff_datetime": "2024-01-01 12:10:00",
        "payment_type": "credit_card",
        "fare_amount": 15.0,
        "tip_amount": 2.0,
    },
    {
        "trip_id": 2,
        "passenger_count": 1,
        "trip_distance": 86.0,
        "pickup_datetime": "2024-01-01 14:00:00",
        "dropoff_datetime": "2024-01-01 15:13:00",
        "payment_type": "cash",
        "fare_amount": 15.0,
        "tip_amount": 3.0,
    },
    {
        "trip_id": 3,
        "passenger_count": 1,
        "trip_distance": 6.0,
        "pickup_datetime": "2024-01-01 15:31:00",
        "dropoff_datetime": "2024-01-01 15:45:00",
        "payment_type": "cash",
        "fare_amount": 15.0,
        "tip_amount": 3.0,
    },
]

response = requests.post(
    "https://<your-app-url>.databricksapps.com/api/v1/table",
    headers={"Authorization": f"Bearer {token}"},
    json={
        "catalog": "my_catalog",
        "schema": "my_schema",
        "table": "trips",
        "data": rows_to_be_inserted,
    },
)
print(response.json())
```

If the request was successful, you will get the following output in your terminal:

```shell
{'data': [{'trip_id': 1, 'passenger_count': 1, 'trip_distance': 10.0, 'pickup_datetime': '2024-01-01 12:00:00', 'dropoff_datetime': '2024-01-01 12:10:00', 'payment_type': 'credit_card', 'fare_amount': 15.0, 'tip_amount': 2.0}, {'trip_id': 2, 'passenger_count': 1, 'trip_distance': 86.0, 'pickup_datetime': '2024-01-01 14:00:00', 'dropoff_datetime': '2024-01-01 15:13:00', 'payment_type': 'cash', 'fare_amount': 15.0, 'tip_amount': 3.0}, {'trip_id': 3, 'passenger_count': 1, 'trip_distance': 6.0, 'pickup_datetime': '2024-01-01 15:31:00', 'dropoff_datetime': '2024-01-01 15:45:00', 'payment_type': 'cash', 'fare_amount': 15.0, 'tip_amount': 3.0}], 'count': 3, 'total': 3}
```

## Resources

- [SQL warehouse](https://docs.databricks.com/aws/en/compute/sql-warehouse/)
- [Unity Catalog table](https://docs.databricks.com/aws/en/tables/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `SELECT` and `MODIFY` on the Unity Catalog table
- `CAN USE` on the SQL warehouse

See Unity [Catalog privileges and securable objects](https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Databricks SQL Connector for Python](https://pypi.org/project/databricks-sql-connector/) - `databricks-sql-connector`
- [FastAPI](https://pypi.org/project/fastapi/) - `fastapi`
- [uvicorn](https://pypi.org/project/uvicorn/) - `uvicorn`

```python title="requirements.txt"
databricks-sdk
databricks-sql-connector
fastapi
uvicorn
```

---

## Read a table


This recipe demonstrates how to query Databricks [Unity Catalog tables](https://docs.databricks.com/aws/en/tables/) from a FastAPI application using the [Databricks SQL Connector](https://docs.databricks.com/en/dev-tools/python-sql-connector.html).

:::info
In this example, we set up our API to be called using the `GET` HTTP method which is the standard choice for reading data in REST APIs as defined in RFC 7231. It's designed to be safe (non-modifying) and idempotent, making it ideal for data retrieval operations.

GET requests are cacheable by default, improving performance, and their parameters are URL-encoded, making them bookmarkable and shareable.

For detailed specifications, refer to [RFC 7231 Section 4.3.1](https://datatracker.ietf.org/doc/html/rfc7231#section-4.3.1) which defines the GET method's semantics and requirements.

:::

## Code snippet

```python title="app.py"

from typing import Dict, List

from fastapi import FastAPI, Query

from databricks import sql
from databricks.sdk.core import Config

DATABRICKS_WAREHOUSE_ID = os.environ.get("DATABRICKS_WAREHOUSE_ID") or None

app = FastAPI()
databricks_cfg = Config()

def get_connection(warehouse_id: str):
    http_path = f"/sql/1.0/warehouses/{DATABRICKS_WAREHOUSE_ID}"
    return sql.connect(
        server_hostname=databricks_cfg.host,
        http_path=http_path,
        credentials_provider=lambda: databricks_cfg.authenticate,
    )

def query(sql_query: str, warehouse_id: str, as_dict: bool = True) -> List[Dict]:
    conn = get_connection(warehouse_id)
    try:
        with conn.cursor() as cursor:
            cursor.execute(sql_query)
            result = cursor.fetchall()
            columns = [col[0] for col in cursor.description]
            return [dict(zip(columns, row)) for row in result]

    except Exception as e:
        raise Exception(f"DBSQL Query Failed: {str(e)}")

@app.get("/api/v1/table")
def table(
    sql_query: str = Query("SELECT * FROM samples.nyctaxi.trips LIMIT 5", description="SQL query to execute"),
):
    results = None
    try:
        results = query(sql_query, warehouse_id=DATABRICKS_WAREHOUSE_ID)
    except Exception as e:
        raise Exception(f"FastAPI Request Failed: {str(e)}")

    return {"results": results}

```

:::warning

The above example is shortened for brevity and not suitable for production use.
You can find a more advanced sample in the databricks-apps-cookbook GitHub repository.

:::

### Example Usage

The query below retrieves data from the [`system.billing.usage` table](https://docs.databricks.com/aws/en/admin/system-tables/billing) via the above code snippet.  
Note that we need to encode the `sql_query` contents when making the API request.

```shell
curl -X GET "https://your-app-url/api/v1/table?sql_query=SELECT%20%2A%20FROM%20system.billing.usage%20LIMIT%201" \
  -H "Authorization: Bearer YOUR_DATABRICKS_TOKEN" | jq
```

```json
{
  "results": [
    {
      "account_id": "abcdef",
      "workspace_id": "12345",
      "record_id": "I_rcFQmY7QFNlZ3nXzNT5uMs",
      "sku_name": "ENTERPRISE_JOBS_SERVERLESS_COMPUTE_US_WEST_OREGON",
      "cloud": "AWS",
      "usage_start_time": "2025-03-20T16:00:00+00:00",
      "usage_end_time": "2025-03-20T17:00:00+00:00",
      "usage_date": "2025-03-20",
      "custom_tags": [],
      "usage_unit": "DBU",
      "usage_quantity": 0.016312123333333334,
      "usage_metadata": {
        "cluster_id": null,
        "job_id": null,
        "warehouse_id": null,
        "instance_pool_id": null,
        "node_type": null,
        "job_run_id": null,
        "notebook_id": null,
        "dlt_pipeline_id": "1f26d5ff-35d2-4d51-b7e7-b64500ec1c6b",
        "endpoint_name": null,
        "endpoint_id": null,
        "dlt_update_id": null,
        "dlt_maintenance_id": null,
        "run_name": null,
        "job_name": null,
        "notebook_path": null,
        "central_clean_room_id": null,
        "source_region": null,
        "destination_region": null,
        "app_id": null,
        "app_name": null,
        "metastore_id": null,
        "private_endpoint_name": null,
        "storage_api_type": null,
        "budget_policy_id": null,
        "ai_runtime_pool_id": null,
        "ai_runtime_workload_id": null
      },
      "identity_metadata": {
        "run_as": "user@example.com",
        "created_by": null,
        "owned_by": null
      },
      "record_type": "ORIGINAL",
      "ingestion_date": "2025-03-20",
      "billing_origin_product": "SQL",
      "product_features": {
        "jobs_tier": null,
        "sql_tier": null,
        "dlt_tier": null,
        "is_serverless": true,
        "is_photon": true,
        "serving_type": null,
        "networking": null,
        "ai_runtime": null
      },
      "usage_type": "COMPUTE_TIME"
    }
  ]
}
```

## Resources

- [SQL warehouse](https://docs.databricks.com/aws/en/compute/sql-warehouse/)
- [Unity Catalog table](https://docs.databricks.com/aws/en/tables/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `SELECT` on the Unity Catalog table
- `CAN USE` on the SQL warehouse

See Unity [Catalog privileges and securable objects](https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Databricks SQL Connector for Python](https://pypi.org/project/databricks-sql-connector/) - `databricks-sql-connector`
- [FastAPI](https://pypi.org/project/fastapi/) - `fastapi`
- [uvicorn](https://pypi.org/project/uvicorn/) - `uvicorn`

```python title="requirements.txt"
databricks-sdk
databricks-sql-connector
fastapi
uvicorn
```

---

## Stream video from volumes


This recipe demonstrates how to stream video files from [Unity Catalog volumes](https://docs.databricks.com/en/volumes/index.html) in a FastAPI application using the [Databricks Files API](https://docs.databricks.com/api/workspace/files) with OAuth authentication.

:::info
This example uses the `GET` HTTP method to stream video content with support for range requests, enabling features like video seeking and progressive loading. The endpoint uses OAuth 2.0 client credentials flow for secure authentication with the Databricks Files API.

Streaming video chunk by chunk ensures efficient memory usage and allows browsers to start playback before the entire file is downloaded.
:::

## Code snippet

```python title="app.py"

from urllib.parse import quote

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import StreamingResponse

app = FastAPI()
logger = logging.getLogger(__name__)

def get_oauth_token() -> str:
    """Get OAuth token using client credentials flow (synchronous)"""
    client_id = os.getenv("DATABRICKS_CLIENT_ID")
    client_secret = os.getenv("DATABRICKS_CLIENT_SECRET")
    
    if not client_id or not client_secret:
        raise HTTPException(
            status_code=500, 
            detail="DATABRICKS_CLIENT_ID and DATABRICKS_CLIENT_SECRET environment variables must be set"
        )
    databricks_host = os.getenv("DATABRICKS_HOST")
    if not databricks_host:
        raise HTTPException(status_code=500, detail="DATABRICKS_HOST environment variable must be set")
    
    token_endpoint = f"https://{databricks_host}/oidc/v1/token"
    
    logger.info(f"Requesting OAuth token from: {token_endpoint}")
    
    try:
        response = requests.post(
            token_endpoint,
            auth=(client_id, client_secret),
            data={
                'grant_type': 'client_credentials',
                'scope': 'all-apis'
            },
            timeout=30
        )
        response.raise_for_status()
        
        token_data = response.json()
        access_token = token_data.get("access_token")
        
        if not access_token:
            logger.error(f"No access_token in response: {token_data}")
            raise HTTPException(status_code=500, detail="Failed to get access token from response")
        
        expires_in = token_data.get("expires_in", 3600)
        logger.info(f"Successfully obtained OAuth token, expires in {expires_in} seconds")
        
        return access_token
    except requests.RequestException as e:
        logger.error(f"OAuth token request failed: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Failed to obtain OAuth token: {str(e)}")

@app.get("/api/files-api/video")
async def stream_video_files_api(request: Request):
    """Stream video from volume using Databricks Files API with OAuth token chunk by chunk"""
    try:
        # Get video path from environment
        file_path = os.getenv("VIDEO_PATH")
        if not file_path:
            raise HTTPException(status_code=400, detail="VIDEO_PATH environment variable not set")
        
        # Get Databricks host
        databricks_host = os.getenv("DATABRICKS_HOST")
        if not databricks_host:
            raise HTTPException(status_code=500, detail="DATABRICKS_HOST environment variable must be set")
        
        # Get OAuth token
        try:
            token = get_oauth_token()
            logger.info(f"Generated OAuth token for Files API streaming")
        except Exception as e:
            logger.error(f"Failed to get OAuth token: {str(e)}")
            raise HTTPException(status_code=401, detail=f"Failed to get OAuth token: {str(e)}")
        
        # URL-encode path, keep slashes
        encoded_path = quote(file_path, safe="/")
        
        # Files API download endpoint: GET /api/2.0/fs/files{file_path}
        url = f"https://{databricks_host}/api/2.0/fs/files{encoded_path}"
        
        logger.info(f"Streaming from Files API URL: {url}")
        
        # Forward Range header for video seeking
        range_header = request.headers.get("Range")
        headers = {
            "Authorization": f"Bearer {token}",
            "Accept": "video/mp4,video/*,*/*",
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        }
        if range_header:
            headers["Range"] = range_header
            logger.info(f"Forwarding Range header: {range_header}")
        
        # Stream the video chunk by chunk using requests with stream=True
        response = requests.get(url, headers=headers, stream=True, timeout=60)
        
        # Log response details for debugging
        logger.info(f"Files API response status: {response.status_code}")
        logger.info(f"Content-Type: {response.headers.get('content-type', 'unknown')}")
        logger.info(f"Content-Length: {response.headers.get('content-length', 'unknown')}")
        
        # Check if we got HTML instead of video (authentication issue)
        content_type = response.headers.get("content-type", "").lower()
        
        # Read first chunk to verify it's not HTML
        first_chunk = b""
        for chunk in response.iter_content(chunk_size=1024):
            first_chunk = chunk
            break
        
        is_html = (
            b"<html" in first_chunk.lower()
            or b"<!doctype html" in first_chunk.lower()
        )
        
        if is_html or "text/html" in content_type:
            logger.error(f"Received HTML response instead of video. Content-Type: {content_type}")
            logger.error(f"First bytes: {first_chunk[:100]}")
            raise HTTPException(
                status_code=401, 
                detail="Files API returned HTML (likely authentication error). Check OAuth token."
            )
        
        response.raise_for_status()
        
        logger.info(f"Successfully validated video stream from Files API")
        
        # Create generator that includes the first chunk we already read
        def generate():
            # Yield the first chunk we already read
            yield first_chunk
            # Then yield the rest
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:
                    yield chunk
        
        return StreamingResponse(
            generate(),
            status_code=response.status_code,
            media_type="video/mp4",
            headers={
                "Accept-Ranges": "bytes",
                "Content-Length": response.headers.get("Content-Length", ""),
                "Content-Range": response.headers.get("Content-Range", "")
            }
        )
        
    except requests.RequestException as e:
        logger.error(f"Files API request error: {str(e)}")
        raise HTTPException(500, f"Network error: {str(e)}")
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Files API proxy error: {str(e)}")
        raise HTTPException(500, f"Proxy error: {str(e)}")
```

:::warning

The above example requires proper OAuth 2.0 credentials in your Databricks App. Ensure the service principal has the necessary permissions to access the volume files.

:::

### Environment Variables

The following environment variables must be set in your app configuration:

- `VIDEO_PATH`: Path to the video file in Unity Catalog volumes (e.g., `/Volumes/catalog/schema/volume/video.mp4`)

## Resources

- [Unity Catalog volumes](https://docs.databricks.com/en/volumes/index.html)
- [Databricks Files API](https://docs.databricks.com/api/workspace/files)
- [OAuth 2.0 authentication](https://docs.databricks.com/en/dev-tools/auth/oauth-m2m.html)
- [FastAPI streaming responses](https://fastapi.tiangolo.com/advanced/custom-response/#streamingresponse)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `READ VOLUME` privilege on the Unity Catalog volume containing the video file

See [Unity Catalog privileges and securable objects](https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges) for more information.

If you declare volume access in a Databricks Asset Bundle, `resources.apps[*].resources[*].uc_securable` may not grant `USE_CATALOG` and `USE_SCHEMA` on the parent catalog and schema (the app still needs them at runtime). As a temporary workaround until bundles can declare those parent grants, add the privileges manually, or see [apps_grants_sync](https://github.com/salihbout/apps_grants_sync): an example Databricks App and Asset Bundle that wires `experimental.scripts.postdeploy` so parent privileges are applied after each `databricks bundle deploy` (copy its `tools/` into your bundle or mirror the same pattern in `databricks.yml`).

## Dependencies

- [FastAPI](https://pypi.org/project/fastapi/) - `fastapi`
- [requests](https://pypi.org/project/requests/) - `requests`
- [uvicorn](https://pypi.org/project/uvicorn/) - `uvicorn`

```python title="requirements.txt"
fastapi
requests
uvicorn
```

---

## Connect Databricks App to Databricks App


This recipe demonstrates how to connect from a Databricks App to an API hosted as another Databricks App.

:::info

This is another type of Machine-to-Machine (M2M) connectivity. As each Databricks App is assigned a dedicated [service principal](https://docs.databricks.com/aws/en/dev-tools/auth/oauth-m2m) upon creation, you should use this service principal's credentials when interacting with any Databricks service, including other Databricks Apps.

:::

## Code snippet

In the Databricks Apps environment, environment variables for `DATABRICKS_HOST`, `DATABRICKS_CLIENT_ID`, and `DATABRICKS_CLIENT_SECRET` are are automatically set based on the app's workspace and assigned service principal. Therefore, you do not need to explicitly specify these values in the following code:

```python
from databricks.sdk import WorkspaceClient

# DATABRICKS_HOST, DATABRICKS_CLIENT_ID, DATABRICKS_CLIENT_SECRET available in environemnt
wc = WorkspaceClient()
headers = wc.config.authenticate() # Contains Bearer Token

response = requests.get(
    "https://your-app-url/api/v1/healthcheck", headers=headers
)
print(response.json())
```

## Permissions

For Databricks App connectivity and authentication, the connecting App [service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN USE` on the target app

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Requests](https://pypi.org/project/requests/) - `requests`

```python title="requirements.txt"
databricks-sdk
requests
```

---

## Connect external app to Databricks App


This recipe demonstrates how to connect from your externally hosted application to an API deployed as a Databricks App.

:::info

This connectivity option is called Machine-to-Machine (M2M) and requires you to set up a Databricks [service principal](https://docs.databricks.com/aws/en/dev-tools/auth/oauth-m2m) as per security best practices.

:::

## Code snippet

The example below requires you to input a valid databricks workspace host URl as well as [service principal](https://docs.databricks.com/aws/en/dev-tools/auth/oauth-m2m) client id and client secret when instantiating the `WorkspaceClient`.

Alternatively, you can set up these values as environment variables `DATABRICKS_HOST`, `DATABRICKS_CLIENT_ID`, and `DATABRICKS_CLIENT_SECRET`. The `WorkspaceClient` will auomatically pick up these environemnt variables.

```python
from databricks.sdk import WorkspaceClient

wc = WorkspaceClient(
    host="https://<DATABRICKS_HOST>",
    client_id="<DATABRICKS_CLIENT_ID>",
    client_secret="<DATABRICKS_CLIENT_SECRET>"
)
headers = wc.config.authenticate() # Contains Bearer Token

response = requests.get(
    "https://your-app-url/api/v1/healthcheck", headers=headers
)
print(response.json())
```

## Permissions

For external app connectivity and authentication, the connecting [service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN USE` on the target app

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Requests](https://pypi.org/project/requests/) - `requests`

```python title="requirements.txt"
databricks-sdk
requests
```

---

## Connect local machine to Databricks App


This recipe demonstrates how to connect from your local machine to an API deployed as a Databricks App.

:::warning

While the FastAPI application running locally on http://127.0.0.1:8000 does not require a [valid bearer token](https:/docs.databricks.com/aws/en/dev-tools/auth/#databricks-authorization-methods), this token is required for accessing Databricks Apps via the secured HTTPS URL with `/api` endpoints.

:::

## Generate OAuth2 Bearer token

The easiest method to generate a valid token is to use the [Databricks CLI](https://docs.databricks.com/aws/en/dev-tools/cli/tutorial) which allows you to authenticate with the workspace where the Databricks App is hosted.

You can log in to a workspace via the following CLI command:

```bash
databricks auth login --host https://<databricks-workspace-url> --profile my-env
```

Upon logging in successfully via the browser, you will see that your profile has been saved. This is helpful if you need to test your app across multiple environments.

```bash
Profile my-env was successfully saved
```

A bearer token can then be generated with a limited time-to-live. Make sure to temporarily store the `access_token` details for use later on.

```bash
databricks auth token --profile my-env
{
  "access_token": "ey....<REDACTED>",
  "token_type": "Bearer",
  "expiry": "2025-04-14T21:11:13.933142+01:00"
}
```

## Code snippets

The following examples below show how the FastAPI application can be called using the bearer token.

#### cURL

```bash
curl -X GET "https://your-app-url/api/v1/healthcheck" \
  -H "Authorization: Bearer YOUR_DATABRICKS_TOKEN"
```

#### Python

```python

response = requests.get(
    "https://your-app-url/api/v1/healthcheck",
    headers={"Authorization": f"Bearer YOUR_DATABRICKS_TOKEN"}
)
print(response.json())
```

If you would like to avoid storing the token in your code, you can leverage the profile you have created using the Databricks CLI through the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/) directly.

```python
from databricks.sdk.core import Config

config = Config(profile="my-env")
token = config.oauth_token().access_token

response = requests.get(
    "https://your-app-url/api/v1/healthcheck",
    headers={"Authorization": f"Bearer {token}"},
)
print(response.json())
```

## Permissions

For local connectivity and authentication, the connecting user(s)/group(s) needs the following permissions:

- `CAN USE` on the target app

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Requests](https://pypi.org/project/requests/) - `requests`

```python title="requirements.txt"
databricks-sdk
requests
```

---

## Connect to a FastAPI app


There are several options to consider when connecting to a deployed Databricks App.

| Option                                                                                            | Use-Case                                                                                                        |
| ------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------- |
| [Local Machine to Databricks App](/docs/fastapi/getting_started/connections/connect_from_local)   | Development & testing, exploratory work                                                                         |
| [External App to Databricks App](/docs/fastapi/getting_started/connections/connect_from_external) | Consume Databricks services from an external application via an API hosted in Databricks Apps                   |
| [Databricks App to Databricks App](/docs/fastapi/getting_started/connections/connect_from_app)    | Internal consumption of Databricks services across workspaces (e.g. app federation), microservices architecture |

---

## Create a FastAPI app


This recipe demonstrates how to create a simple [FastAPI](https://fastapi.tiangolo.com/) application that can be deployed on Databricks Apps.

## Code structure

Our FastAPI application will initially have the following structure:

| File                       | Purpose                                                            |
| -------------------------- | ------------------------------------------------------------------ |
| `app.py`                   | Main FastAPI application entrypoint                                |
| `app.yaml`                 | Databricks Apps deployment configuration                           |
| `requirements.txt`         | Main application dependencies                                      |
| `routes/__init__.py`       | Module initialisation for API-router                               |
| `routes/v1/__init__.py`    | Sub-module initialisation for API-router (e.g. `/api/v1/*` routes) |
| `routes/v1/healthcheck.py` | Isolated code handler for healthcheck                              |

```bash
tree -L 3
.
├── app.py
├── app.yaml
├── routes
│   ├── __init__.py
│   └── v1
│       ├── __init__.py
│       └── healthcheck.py
├── requirements.txt
```

### Main application entrypoint

```python title="app.py"
from fastapi import FastAPI

from routes import api_router

app = FastAPI(
    title="FastAPI & Databricks Apps",
    description="A simple FastAPI application example for Databricks Apps runtime",
    version="1.0.0",
)

# Router assignment
app.include_router(api_router)
```

### API routes

:::warning

In order to use OAuth2 Bearer token authentication with Databricks Apps, your application code must provide valid routes with a prefix of `/api`.

:::

Let's define the main router entrypoint for the application.

```python title="routes/__init__.py"
from fastapi import APIRouter

# Import routers from versioned packages
from .v1 import router as v1_router

# Create a router for the API
api_router = APIRouter()

# Include versioned routers - prefix must have /api for Databricks Apps token-based auth
api_router.include_router(v1_router, prefix="/api/v1")
```

We then can define specific routers for the application, using versioning and specific endpoints for isolation per optimal practice.

Next, we can create a dedicated file to handle healthchecks, exposed via `https://<app-url>/api/v1/healthcheck`.

```python title="routes/v1/__init__.py"
"""V1 API routes."""

from fastapi import APIRouter

from .healthcheck import router as healthcheck_router

router = APIRouter()

# Include endpoint-specific routers
router.include_router(healthcheck_router)
```

```python title="routes/v1/healthcheck.py"
from datetime import datetime, timezone
from fastapi import APIRouter
from typing import Dict

router = APIRouter()

@router.get("/healthcheck")
async def healthcheck() -> Dict[str, str]:
    """Return the API status."""
    return {"status": "OK", "timestamp": datetime.now(timezone.utc).isoformat()}
```

## Running the FastAPI application

### Local machine

To run the application locally, use the following terminal command: `uvicorn app:app --reload`:

```bash
INFO:     Will watch for changes in these directories: ['/Users/user.name/home/databricks-apps-cookbook/fastapi']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [56162] using StatReload
INFO:     Started server process [56164]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
```

Once the app is running, you will be to access the following endpoints:

- Index Page: http://127.0.0.1:8000
- OpenAPI Docs: http://127.0.0.1:8000/docs
- Healthcheck: http://127.0.0.1:8000/api/v1/healthcheck

Accessing the app or making changes to `app.py` will update the tailing application log in the terminal:

```bash
INFO:     127.0.0.1:54508 - "GET / HTTP/1.1" 200 OK
INFO:     127.0.0.1:58696 - "GET /docs HTTP/1.1" 200 OK
INFO:     127.0.0.1:58696 - "GET /openapi.json HTTP/1.1" 200 OK
INFO:     127.0.0.1:58893 - "GET /api/v1/healthcheck HTTP/1.1" 200 OK
```

```bash
WARNING:  StatReload detected changes in 'app.py'. Reloading...
INFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
INFO:     Finished server process [56164]
INFO:     Started server process [63412]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
```

### Databricks Apps runtime

Assuming the code works locally, you can also deploy this application to a Databricks workspace using Databaricks Apps.

Follow the [Get started with Databricks Apps](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/get-started) for deployment instructions.

For the app [configuration](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/configuration#configuring-your-databricks-apps-with-the-appyaml-file), use:

```yaml title="app.yaml"
command: ["uvicorn", "app:app"]
```

The [FastAPI](https://pypi.org/project/fastapi/) and [uvicorn](https://pypi.org/project/uvicorn/) packages are included in the [default Databricks Apps environment](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#installed-python-libraries).

For parity with your local code and to trigger Databricks Apps to install the latest package version, include a `requirements.txt` in your application folder as described in the following section.

## Dependencies

- [FastAPI](https://pypi.org/project/fastapi/) - `fastapi`
- [uvicorn](https://pypi.org/project/uvicorn/) - `uvicorn`

```python title="requirements.txt"
fastapi
uvicorn
```

---

## Connect FastAPI to Lakebase


This guide demonstrates how to connect your FastAPI application to a Lakebase PostgreSQL database with automatic token refresh and connection pooling. Lakebase provides a managed PostgreSQL database instance within your Databricks workspace that seamlessly integrates with Unity Catalog.

:::info What is Lakebase?
Lakebase is Databricks' managed PostgreSQL service that provides:
- **Fully managed PostgreSQL instances** within your Databricks workspace
- **Automatic OAuth token authentication** using Databricks credentials
- **Unity Catalog integration** for unified data governance
- **High availability and automatic backups**
- **Seamless data synchronization** with Unity Catalog tables
:::

## Prerequisites

Before connecting to Lakebase, ensure you have:
- A Lakebase PostgreSQL instance created (see [Create Lakebase Resources](../building_endpoints/lakebase/lakebase_resources_create.mdx))
- FastAPI application with required dependencies installed
- Databricks workspace authentication configured locally

## Database Configuration

### Environment Variables

```bash title=".env"
# Lakebase Configuration
LAKEBASE_INSTANCE_NAME=my-lakebase-instance
LAKEBASE_DATABASE_NAME=my_database
LAKEBASE_CATALOG_NAME=my-pg-catalog

# Database Connection Pool Settings
DB_POOL_SIZE=5
DB_MAX_OVERFLOW=10
DB_COMMAND_TIMEOUT=30
DB_POOL_TIMEOUT=10
DB_POOL_RECYCLE_INTERVAL=3600
DATABRICKS_DATABASE_PORT=5432
```

### Complete Implementation

```python title="config/database.py"

from typing import AsyncGenerator

from databricks.sdk import WorkspaceClient
from dotenv import load_dotenv
from sqlalchemy import URL, event, text
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

load_dotenv()
logger = logging.getLogger(__name__)

# Global variables
engine: AsyncEngine | None = None
AsyncSessionLocal: sessionmaker | None = None
workspace_client: WorkspaceClient | None = None
database_instance = None

# Token management for background refresh
postgres_password: str | None = None
last_password_refresh: float = 0
token_refresh_task: asyncio.Task | None = None

async def refresh_token_background():
    """Background task to refresh tokens every 50 minutes"""
    global postgres_password, last_password_refresh, workspace_client, database_instance

    while True:
        try:
            await asyncio.sleep(50 * 60)  # Wait 50 minutes
            logger.info("Background token refresh: Generating fresh PostgreSQL OAuth token")

            cred = workspace_client.database.generate_database_credential(
                request_id=str(uuid.uuid4()),
                instance_names=[database_instance.name],
            )
            postgres_password = cred.token
            last_password_refresh = time.time()
            logger.info("Background token refresh: Token updated successfully")

        except Exception as e:
            logger.error(f"Background token refresh failed: {e}")

def init_engine():
    """Initialize database connection using SQLAlchemy with automatic token refresh"""
    global engine, AsyncSessionLocal, workspace_client, database_instance, postgres_password, last_password_refresh

    try:
        # Initialize Databricks SDK client
        workspace_client = WorkspaceClient()

        # Get database instance details
        instance_name = os.getenv("LAKEBASE_INSTANCE_NAME")
        if not instance_name:
            raise RuntimeError("LAKEBASE_INSTANCE_NAME environment variable is required")
            
        database_instance = workspace_client.database.get_database_instance(name=instance_name)

        # Generate initial OAuth credentials
        cred = workspace_client.database.generate_database_credential(
            request_id=str(uuid.uuid4()), 
            instance_names=[database_instance.name]
        )
        postgres_password = cred.token
        last_password_refresh = time.time()
        logger.info("Database: Initial credentials generated")

        # Build connection URL
        database_name = os.getenv("LAKEBASE_DATABASE_NAME", database_instance.name)
        username = (
            os.getenv("DATABRICKS_CLIENT_ID")
            or workspace_client.current_user.me().user_name
            or None
        )

        url = URL.create(
            drivername="postgresql+asyncpg",
            username=username,
            password="",  # Will be set by event handler
            host=database_instance.read_write_dns,
            port=int(os.getenv("DATABRICKS_DATABASE_PORT", "5432")),
            database=database_name,
        )

        # Create async engine with connection pooling
        engine = create_async_engine(
            url,
            pool_pre_ping=False,
            echo=False,
            pool_size=int(os.getenv("DB_POOL_SIZE", "5")),
            max_overflow=int(os.getenv("DB_MAX_OVERFLOW", "10")),
            pool_timeout=int(os.getenv("DB_POOL_TIMEOUT", "10")),
            pool_recycle=int(os.getenv("DB_POOL_RECYCLE_INTERVAL", "3600")),
            connect_args={
                "command_timeout": int(os.getenv("DB_COMMAND_TIMEOUT", "30")),
                "server_settings": {
                    "application_name": "fastapi_orders_app",
                },
                "ssl": "require",
            },
        )

        # Register token provider for new connections
        @event.listens_for(engine.sync_engine, "do_connect")
        def provide_token(dialect, conn_rec, cargs, cparams):
            global postgres_password
            # Use current token from background refresh
            cparams["password"] = postgres_password

        # Create session factory
        AsyncSessionLocal = sessionmaker(
            bind=engine, 
            class_=AsyncSession, 
            expire_on_commit=False
        )
        
        logger.info(f"Database engine initialized for {database_name} with background token refresh")

    except Exception as e:
        logger.error(f"Error initializing database: {e}")
        raise RuntimeError(f"Failed to initialize database: {e}") from e

async def start_token_refresh():
    """Start the background token refresh task"""
    global token_refresh_task
    if token_refresh_task is None or token_refresh_task.done():
        token_refresh_task = asyncio.create_task(refresh_token_background())
        logger.info("Background token refresh task started")

async def stop_token_refresh():
    """Stop the background token refresh task"""
    global token_refresh_task
    if token_refresh_task and not token_refresh_task.done():
        token_refresh_task.cancel()
        try:
            await token_refresh_task
        except asyncio.CancelledError:
            pass
        logger.info("Background token refresh task stopped")

async def get_async_db() -> AsyncGenerator[AsyncSession, None]:
    """Get a database session with automatic token refresh"""
    if AsyncSessionLocal is None:
        raise RuntimeError("Engine not initialized; call init_engine() first")
    async with AsyncSessionLocal() as session:
        yield session

def check_database_exists() -> bool:
    """Check if the Lakebase database instance exists"""
    try:
        workspace_client = WorkspaceClient()
        instance_name = os.getenv("LAKEBASE_INSTANCE_NAME")
        
        if not instance_name:
            logger.warning("LAKEBASE_INSTANCE_NAME not set - database instance check skipped")
            return False
            
        workspace_client.database.get_database_instance(name=instance_name)
        logger.info(f"Lakebase database instance '{instance_name}' exists")
        return True
    except Exception as e:
        if "not found" in str(e).lower() or "resource not found" in str(e).lower():
            logger.info(f"Lakebase database instance '{instance_name}' does not exist")
        else:
            logger.error(f"Error checking database instance existence: {e}")
        return False

async def database_health() -> bool:
    """Check database connection health"""
    global engine

    if engine is None:
        logger.error("Database engine failed to initialize.")
        return False

    try:
        async with engine.connect() as connection:
            await connection.execute(text("SELECT 1"))
            logger.info("Database connection is healthy.")
            return True
    except Exception as e:
        logger.error("Database health check failed: %s", e)
        return False
```

## Integration with FastAPI

### Application Startup

```python title="main.py"
from contextlib import asynccontextmanager
from fastapi import FastAPI
from config.database import init_engine, start_token_refresh, stop_token_refresh, check_database_exists

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan management"""
    # Startup
    if check_database_exists():
        init_engine()
        await start_token_refresh()
        logger.info("Application started with Lakebase connection")
    else:
        logger.warning("Lakebase database not found - orders endpoints disabled")
    
    yield
    
    # Shutdown
    await stop_token_refresh()
    logger.info("Application shutdown complete")

app = FastAPI(lifespan=lifespan)
```

### Using Database Sessions

```python title="routes/orders.py"
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from config.database import get_async_db
from models.orders import Order

router = APIRouter()

@router.get("/orders/count")
async def get_order_count(db: AsyncSession = Depends(get_async_db)):
    """Get total orders count using dependency injection"""
    stmt = select(func.count(Order.o_orderkey))
    result = await db.execute(stmt)
    count = result.scalar()
    return {"total_orders": count}
```

## Key Architecture Decisions

### 1. Automatic Token Refresh
```python
# OAuth tokens expire after 1 hour
# We refresh every 50 minutes to ensure continuous connectivity
await asyncio.sleep(50 * 60)  # 50 minutes
```

### 2. Connection Event Handler
```python
# SQLAlchemy event listener injects fresh tokens for new connections
@event.listens_for(engine.sync_engine, "do_connect")
def provide_token(dialect, conn_rec, cargs, cparams):
    cparams["password"] = postgres_password
```

### 3. Connection Pooling Configuration
```python
engine = create_async_engine(
    url,
    pool_size=5,           # Base number of connections
    max_overflow=10,       # Additional connections under load
    pool_timeout=10,       # Max wait time for connection
    pool_recycle=3600,     # Recycle connections every hour
)
```

## Configuration Options

### Connection Pool Settings

| Parameter | Default | Description | Recommended |
|-----------|---------|-------------|-------------|
| `DB_POOL_SIZE` | 5 | Base connection pool size | 5-10 for most apps |
| `DB_MAX_OVERFLOW` | 10 | Additional connections under load | 2x pool_size |
| `DB_POOL_TIMEOUT` | 10 | Max seconds to wait for connection | 10-30 seconds |
| `DB_COMMAND_TIMEOUT` | 30 | Query timeout in seconds | 30-60 seconds |
| `DB_POOL_RECYCLE_INTERVAL` | 3600 | Recycle connections (seconds) | 3600 (1 hour) |

### Performance Tuning

```python
# For high-traffic applications
DB_POOL_SIZE=10
DB_MAX_OVERFLOW=20
DB_POOL_TIMEOUT=30

# For development/testing
DB_POOL_SIZE=2
DB_MAX_OVERFLOW=5
DB_POOL_TIMEOUT=10
```

## Health Monitoring

### Database Health Check Endpoint

```python title="routes/health.py"
from fastapi import APIRouter
from config.database import database_health, check_database_exists

router = APIRouter()

@router.get("/health/database")
async def health_check():
    """Comprehensive database health check"""
    instance_exists = check_database_exists()
    connection_healthy = await database_health() if instance_exists else False
    
    return {
        "database_instance_exists": instance_exists,
        "connection_healthy": connection_healthy,
        "status": "healthy" if (instance_exists and connection_healthy) else "unhealthy"
    }
```

## Error Handling and Troubleshooting

### Common Issues

#### 1. Token Refresh Failures
```python
# Monitor token refresh in logs
logger.error(f"Background token refresh failed: {e}")

# Check Databricks SDK authentication
workspace_client = WorkspaceClient()
print(workspace_client.current_user.me())
```

#### 2. Connection Pool Exhaustion
```bash
# Increase pool settings
DB_POOL_SIZE=10
DB_MAX_OVERFLOW=20
DB_POOL_TIMEOUT=30
```

#### 3. Instance Not Found
```bash
# Verify instance name matches exactly
LAKEBASE_INSTANCE_NAME=your-exact-instance-name

# Check instance exists in Databricks workspace
```

## Security Considerations

1. **OAuth Token Security**: Tokens are stored in memory only, never persisted to disk
2. **SSL Enforcement**: All connections require SSL encryption
3. **Connection Isolation**: Each request gets its own database session
4. **Automatic Cleanup**: Sessions are automatically closed after request completion
5. **No Hardcoded Credentials**: All authentication uses Databricks SDK

## Related Documentation

- [Create Lakebase Resources](../building_endpoints/lakebase/lakebase_resources_create.mdx) - Set up your Lakebase instance
- [Lakebase Orders Management](../building_endpoints/lakebase/lakebase_orders.mdx) - Use the database connection
- [Delete Lakebase Resources](../building_endpoints/lakebase/lakebase_resources_delete.mdx) - Clean up resources

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [SQLAlchemy](https://pypi.org/project/sqlalchemy/) - `sqlalchemy`
- [asyncpg](https://pypi.org/project/asyncpg/) - `asyncpg`
- [python-dotenv](https://pypi.org/project/python-dotenv/) - `python-dotenv`

```python title="requirements.txt"
databricks-sdk>=0.60.0
sqlalchemy
asyncpg
python-dotenv
```

---

## Test a FastAPI app


This recipe demonstrates how to test [FastAPI](https://fastapi.tiangolo.com/) applications using [Pytest](https://docs.pytest.org/en/stable/index.html).

Testing is crucial in API development to ensure reliability, maintainability, and confidence in your codebase.

Using Pytest with [dependency injection (via fixtures)](https://docs.pytest.org/en/4.6.x/fixture.html) provides several benefits:

- Allows you to isolate components, making tests more focused and easier to debug.
- Enables you to mock external dependencies (like databases or third-party services) to test your API logic in isolation.
- Helps maintain test consistency by reusing setup code across multiple tests.

This approach not only catches bugs early but also makes your code more modular and easier to refactor, as changes in one component can be tested without affecting others.

Additionally, well-structured tests serve as living documentation, helping new developers understand how different parts of your API are expected to work.

## Code structure

To test our FastAPI application, we will augment the existing code structure to include the following:

| File                                  | Purpose                                                                 |
| ------------------------------------- | ----------------------------------------------------------------------- |
| `pytest.ini`                          | Pytest configuration                                                    |
| `tests/conftest.py`                   | Root Pytest behaviours (e.g. global fixtures)                           |
| `tests/test_app.py`                   | Test suite for main application                                         |
| `tests/routes/v1/test_healthcheck.py` | Test suite for healthcheck endpoint                                     |
| `tests/**/__init__.py`                | Dummy file to register folders as modules under respective test folders |

```bash
tree -L 4
.
├── pytest.ini
├── tests
│   ├── __init__.py
│   ├── conftest.py
│   ├── routes
│   │   ├── __init__.py
│   │   └── v1
│   │       ├── __init__.py
│   │       └── test_healthcheck.py
│   └── test_app.py
```

### Pytest setup

```ini title="pytest.ini"
[pytest]
testpaths = tests
python_files = test_*.py
python_functions = test_*
python_classes = Test*
addopts = -v
```

```python title="tests/conftest.py"

from fastapi.testclient import TestClient

from app import app

@pytest.fixture
def client():
    """Create a test client for the FastAPI application."""
    return TestClient(app)
```

### Tests per FastAPI code

```python title="tests/test_app.py"
from fastapi import status

def test_root_endpoint(client):
    """Test the root endpoint returns the expected response."""
    response = client.get("/")
    assert response.status_code == status.HTTP_200_OK
    data = response.json()
    assert "app" in data
    assert "message" in data
    assert data["app"] == "Databricks FastAPI Example"
```

```python title="tests/routes/v1/test_healthcheck.py"
from fastapi import status

def test_healthcheck(client):
    """Test the healthcheck endpoint."""
    response = client.get("/api/v1/healthcheck")
    assert response.status_code == status.HTTP_200_OK
    data = response.json()
    assert "status" in data
    assert data["status"] == "OK"
    assert "timestamp" in data
```

## Running tests

:::info

The tests we have included do not require the FastAPI application to be running, but we provide examples of [dependency injection](https://docs.pytest.org/en/4.6.x/fixture.html#fixtures-a-prime-example-of-dependency-injection) in the cookbook source code when not connected to a live environment.

:::

```bash
# Run all tests
pytest

# Run specific tests
pytest tests/routes/v1/test_healthcheck.py

# Run with coverage report
pytest --cov=app tests/
```

## Dependencies

- [FastAPI](https://pypi.org/project/fastapi/) - `fastapi`
- [pytest](https://pypi.org/project/pytest/) - `pytest`
- [pytest-cov](https://pypi.org/project/pytest-cov/) - `pytest-cov`
- [uvicorn](https://pypi.org/project/uvicorn/) - `uvicorn`

```python title="requirements.txt"
fastapi
pytest
pytest-cov
uvicorn
```

---

## Introduction


Welcome to the Databricks Apps Cookbook!

The Databricks Apps Cookbook contains ready-to-use code snippets for building interactive data and AI applications using [Databricks Apps](https://docs.databricks.com/en/dev-tools/databricks-apps/index.html).

These code snippets cover common use cases such as reading and writing to and from **tables** and **volumes**, invoking traditional **ML models** and GenAI, or triggering **workflows**.

For each snippet, you will find the **source code**, required **permissions**, list of **dependencies**, and any other information needed to implement it.

Currently, we offer snippets for [Streamlit](/docs/category/streamlit), [Dash](/docs/category/dash), [FastAPI](/docs/category/fastapi), and [Reflex](/docs/category/reflex) and they can be easily adapted to other Python frameworks.

    <Link to="/docs/category/streamlit" className="w-full sm:w-auto">
        <button className="inline-flex h-14 w-full items-center justify-center gap-3 border-2 border-lava-600 bg-transparent px-6 py-2.5 text-lg font-semibold text-lava-600 transition-colors hover:border-lava-700 hover:bg-lava-600 hover:text-white">
            
            Build with Streamlit
        </button>
    </Link>
    <Link to="/docs/category/dash" className="w-full sm:w-auto">
        <button className="inline-flex h-14 w-full items-center justify-center gap-3 border-2 border-lava-600 bg-transparent px-6 py-2.5 text-lg font-semibold text-lava-600 transition-colors hover:border-lava-700 hover:bg-lava-600 hover:text-white">
            
            Build with Dash
        </button>
    </Link>
    <Link to="/docs/category/fastapi" className="w-full sm:w-auto">
        <button className="inline-flex h-14 w-full items-center justify-center gap-3 border-2 border-lava-600 bg-transparent px-6 py-2.5 text-lg font-semibold text-lava-600 transition-colors hover:border-lava-700 hover:bg-lava-600 hover:text-white">
            
            Build with FastAPI
        </button>
    </Link>
    <Link to="/docs/category/reflex" className="w-full sm:w-auto">
        <button className="inline-flex h-14 w-full items-center justify-center gap-3 border-2 border-lava-600 bg-transparent px-6 py-2.5 text-lg font-semibold text-lava-600 transition-colors hover:border-lava-700 hover:bg-lava-600 hover:text-white">
            
            Build with Reflex
        </button>
    </Link>

## Interactive samples

You can find **interactive sample implementations** for each snippet in the [databricks-apps-cookbook](https://github.com/databricks-solutions/databricks-apps-cookbook) GitHub repository. Take a look at the [deployment instructions](/docs/deploy) to **run them locally** or **deploy to your Databricks workspace**.

![Example banner](./assets/demo.gif)

## Contributing

We welcome contributions! Submit a [pull request](https://github.com/databricks-solutions/databricks-apps-cookbook/pulls) to add or improve recipes. Raise an [issue](https://github.com/databricks-solutions/databricks-apps-cookbook/issues) to report a bug or raise a feature request.

---

## Connect an MCP server(Aiml)


This recipe connects to an [MCP](https://modelcontextprotocol.io/overview) server for AI applications using GitHub as an example and Unity Catalog [HTTP connections](https://docs.databricks.com/aws/en/query-federation/http) for secure and governed access.

## Code snippets

### OAuth User to Machine Per User (On-behalf-of-user)

```python title="app.py"

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ExternalFunctionRequestHttpMethod

def get_client_obo(headers) -> WorkspaceClient:
    """Build a WorkspaceClient using the OBO token forwarded by Databricks Apps."""
    token = getattr(headers, "x_forwarded_access_token", "")
    host = getattr(headers, "x_forwarded_host", "")
    return WorkspaceClient(host=host, token=token, auth_type="pat")

def init_mcp_session(w: WorkspaceClient, connection_name: str):
    init_payload = {
        "jsonrpc": "2.0",
        "id": "init-1",
        "method": "initialize",
        "params": {}
    }
    response = w.serving_endpoints.http_request(
        conn=connection_name,
        method=ExternalFunctionRequestHttpMethod.POST,
        path="/",
        json=init_payload,
    )
    return (response.headers.get("mcp-session-id") or response.headers.get("Mcp-Session-Id"))

class McpState(rx.State):
    response_data: str = ""
    error_message: str = ""

    @rx.event(background=True)
    async def send_request(self):
        async with self:
            self.response_data = ""
            self.error_message = ""
            headers = self.router.headers

        w = get_client_obo(headers)
        if not getattr(headers, "x_forwarded_access_token", ""):
            async with self:
                self.error_message = (
                    "No OBO token found. Enable on-behalf-of-user authentication for this Databricks App."
                )
            return

        connection_name = "github_u2m_connection"
        http_method = ExternalFunctionRequestHttpMethod.POST
        path = "/"
        req_headers = {"Content-Type": "application/json"}
        payload = {"jsonrpc": "2.0", "id": "list-1", "method": "tools/list"}

        session_id = init_mcp_session(w, connection_name)
        if session_id:
            req_headers["Mcp-Session-Id"] = session_id

        response = w.serving_endpoints.http_request(
            conn=connection_name,
            method=http_method,
            path=path,
            headers=req_headers,
            json=payload,
        )
        resp_data = response.as_dict() if hasattr(response, "as_dict") else response
        async with self:
            self.response_data = json.dumps(resp_data, indent=2)
```

### Bearer token

```python title="app.py"
# Reflex (Databricks) version for a simple GET using a bearer/token-auth HTTP connection.

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ExternalFunctionRequestHttpMethod

w = WorkspaceClient()

response = w.serving_endpoints.http_request(
    conn="github_u2m_connection",
    method=ExternalFunctionRequestHttpMethod.GET,
    path="/",
    headers={"Accept": "application/vnd.github+json"},
    json={
        "jsonrpc": "2.0",
        "id": "init-1",
        "method": "initialize",
        "params": {}
    },
)

rx.code_block(json.dumps(response.as_dict(), indent=2), language="json")
```

## Resources

- [Unity Catalog HTTP Connection](https://docs.databricks.com/aws/en/query-federation/http) with the MCP (/mcp) base path

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CONNECTION` permission on the HTTP Connection

When using OAuth User to Machine Per User (On-behalf-of-user), you need to configure [User authorization](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/auth#user-authorization) by adding the Unity Catalog connection or other scopes.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`
- [MCP CLI](https://pypi.org/project/mcp/) - `mcp[cli]`

```python title="requirements.txt"
databricks-sdk
reflex
mcp[cli]
```

---

## Invoke a model(Aiml)


This recipe invokes a model hosted on [Mosaic AI Model Serving](https://docs.databricks.com/aws/en/machine-learning/model-serving/) using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/) and returns the result. Choose either a traditional ML model or a large language model (LLM).

## Code snippets

### Traditional Machine Learning

#### Using `dataframe_split` (JSON-serialized DataFrame in split orientation)

```python title="app.py"
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import DataframeSplitInput

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="traditional-model",
    dataframe_split={
        "columns": ["feature1", "feature2"],
        "data": [[1, 2], [3, 4]]
    }
)
rx.text(response.as_dict())
```

#### Using `dataframe_records` (JSON-serialized DataFrame in records orientation)

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="traditional-model",
    dataframe_records=[
        {"feature1": 1, "feature2": 2},
        {"feature1": 3, "feature2": 4}
    ]
)
rx.text(response.as_dict())
```

#### Using `instances` (Tensor inputs in row format for TensorFlow/PyTorch models)

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="tf-model",
    instances=[
        [1, 2], [3, 4]
    ]
)
rx.text(response.as_dict())
```

#### Using `inputs` (Tensor inputs in columnar format for TensorFlow/PyTorch models)

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="pytorch-model",
    inputs={"input_ids": [1, 2, 3]}
)
rx.text(response.as_dict())
```

### Large language models (LLMs)

#### Using `prompt` (Input text for completion tasks)

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="completion-model",
    prompt="Once upon a time"
)
rx.text(response.as_dict())
```

#### Using `messages` (List of chat messages for conversational models)

```python title="app.py"
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ChatMessage, ChatMessageRole

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="chat-model",
    messages=[
        {"role": "user", "content": "Hello!"},
        {"role": "assistant", "content": "Hi there!"}
    ]
)
rx.text(response.as_dict())
```

#### Using `input` (Input text for embedding tasks)

```python title="app.py"
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="embeddings-model",
    input=["text to embed"]
)
rx.text(response.as_dict())
```

## Resources

- [Model Serving endpoint](https://docs.databricks.com/aws/en/machine-learning/model-serving/manage-serving-endpoints)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN QUERY` on the model serving endpoint

See [Manage permissions on your model serving endpoint](https://docs.databricks.com/aws/en/machine-learning/model-serving/manage-serving-endpoints#manage-permissions-on-your-model-serving-endpoint) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`

```python title="requirements.txt"
databricks-sdk
reflex
```

---

## Run vector search(Aiml)


This recipe performs a vector search query on a [Mosaic AI Vector Search](https://docs.databricks.com/aws/en/generative-ai/vector-search) index using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/).

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()
openai_client = w.serving_endpoints.get_open_ai_client()
EMBEDDING_MODEL_ENDPOINT_NAME = "databricks-gte-large-en"

def get_embeddings(text: str):
    try:
        response = openai_client.embeddings.create(
            model=EMBEDDING_MODEL_ENDPOINT_NAME, input=text
        )
        return response.data[0].embedding
    except Exception as e:
        return f"Error generating embeddings: {e}"

def run_vector_search(index_name: str, columns: str, prompt: str):
    columns_list = [col.strip() for col in columns.split(",") if col.strip()]
    prompt_vector = get_embeddings(prompt)

    if prompt_vector is None or isinstance(prompt_vector, str):
        return str(prompt_vector)

    try:
        results = w.vector_search_indexes.query_index(
            index_name=index_name,
            columns=columns_list,
            query_vector=prompt_vector,
            num_results=3
        )
        return results.result.data_array
    except Exception as e:
        return f"Error running vector search: {e}"

class VectorSearchState(rx.State):
    index_name: str = ""
    columns: str = ""
    search_query: str = ""
    search_results: str = ""
    is_searching: bool = False

    async def perform_search(self):
        self.is_searching = True
        yield
        result = run_vector_search(self.index_name, self.columns, self.search_query)
        self.search_results = str(result)
        self.is_searching = False
```

## Resources

- [Vector Search endpoint](https://docs.databricks.com/aws/en/generative-ai/vector-search)
- [Vector Search index](https://docs.databricks.com/aws/en/generative-ai/vector-search)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CATALOG` on the catalog that contains the Vector Search index
- `USE SCHEMA` on the schema that contains the Vector Search index
- `SELECT` on the Vector Search index

See [Query a vector search endpoint](https://docs.databricks.com/aws/en/generative-ai/vector-search) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`

```python title="requirements.txt"
databricks-sdk
reflex
```

---

## Retrieve a secret(Authentication)


This recipe retrieves a [Databricks secret](https://docs.databricks.com/en/security/secrets/index.html). Use secrets to securely connect to external services and APIs.

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

def get_secret(scope, key):
    try:
        secret_response = w.secrets.get_secret(scope=scope, key=key)
        decoded_secret = base64.b64decode(secret_response.value).decode('utf-8')
        return decoded_secret
    except Exception as e:
        return None

class RetrieveASecretState(rx.State):
    scope_name: str = "my_secret_scope"
    secret_key: str = "api_key"
    is_loading: bool = False
    error_message: str = ""
    success_message: str = ""

    @rx.event(background=True)
    async def retrieve_secret(self):
        async with self:
            self.is_loading = True
            self.error_message = ""
            self.success_message = ""

        try:
            secret = get_secret(self.scope_name, self.secret_key)
            async with self:
                if secret:
                    self.success_message = "Secret retrieved! The value is securely handled in the backend."
                else:
                    self.error_message = "Secret not found or inaccessible. Please create a secret scope and key before retrieving."
        except Exception:
            async with self:
                self.error_message = "Secret not found or inaccessible. Please create a secret scope and key before retrieving."
        finally:
            async with self:
                self.is_loading = False
```

## Resources

- [Secret scope and secret](https://docs.databricks.com/aws/en/security/secrets/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN READ` on the secret scope

See [Manage secret scope permissions](https://docs.databricks.com/aws/en/security/secrets/#manage-secret-scope-permissions) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`

```python title="requirements.txt"
databricks-sdk
reflex
```

---

## Get current user(Authentication)


This recipe gets information about the user accessing this Databricks App from their [HTTP headers](https://docs.databricks.com/en/dev-tools/databricks-apps/app-development.html#what-http-headers-are-passed-to-databricks-apps).

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

class GetCurrentUserState(rx.State):
    header_email: str = ""
    header_username: str = ""
    header_user: str = ""
    header_ip: str = ""
    header_token: str = ""
    user_id: str = ""
    user_name: str = ""
    user_display_name: str = ""
    user_active: bool = False
    user_external_id: str = ""
    groups_count: int = 0
    entitlements_count: int = 0
    all_headers_json: str = ""
    user_json: str = ""
    is_loading: bool = False
    error_message: str = ""

    @rx.event(background=True)
    async def get_user_headers(self):
        async with self:
            self.is_loading = True
            self.error_message = ""
            headers = self.router.headers
            # Convert HeaderData to dict safely
            try:
                headers_dict = {k: v for k, v in vars(headers).items() if not k.startswith('_')}
            except Exception:
                headers_dict = {}
            self.all_headers_json = json.dumps(headers_dict, indent=2)

            self.header_email = getattr(headers, "x_forwarded_email", "")
            self.header_username = getattr(
                headers, "x_forwarded_preferred_username", ""
            )
            self.header_user = getattr(headers, "x_forwarded_user", "")
            self.header_ip = getattr(headers, "x_real_ip", "")
            self.header_token = getattr(headers, "x_forwarded_access_token", "")
        try:
            if self.header_token:
                loop = asyncio.get_running_loop()

                def fetch_user_info(token):
                    w = WorkspaceClient(token=token, auth_type="pat")
                    return w.current_user.me()

                me = await loop.run_in_executor(
                    None, fetch_user_info, self.header_token
                )
                async with self:
                    self.user_id = me.id
                    self.user_name = me.user_name or ""
                    self.user_display_name = me.display_name or ""
                    self.user_active = me.active
                    self.user_external_id = me.external_id or ""
                    self.groups_count = len(me.groups) if me.groups else 0
                    self.entitlements_count = len(me.entitlements) if me.entitlements else 0
                    self.user_json = json.dumps(me.as_dict(), indent=2)
            else:
                async with self:
                    pass
        except Exception as e:
            logging.exception(f"Error fetching current user: {e}")
            async with self:
                self.error_message = f"Error fetching user info: {e}"
        finally:
            async with self:
                self.is_loading = False
```

:::info
This sample requires on-behalf-of-user authentication to be enabled for your app to access the X-Forwarded-Access-Token header. Without this, you will only have access to basic user information from the headers, not the detailed information from the Databricks API. Without the user token present, w.current_user.me() will return information about the app service principal.
:::

## Resources

No Databricks resources are required for this recipe.

## Permissions

No permissions configuration required for accessing headers. To use the `current_user.me()` API, the app must be configured with [on-behalf-of-user authentication](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/app-development#-using-the-databricks-apps-authorization-model).

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`

```python title="requirements.txt"
databricks-sdk
reflex
```

---

## On-behalf-of-user authentication


This recipe demonstrates how to use Databricks Apps [on-behalf-of-user authentication](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/app-development#-using-the-databricks-apps-authorization-model) to run a SQL query using the user's credentials instead of the app's service principal.

## Code snippet

```python title="app.py"

from databricks import sql
from databricks.sdk.core import Config

cfg = Config()

_connection = None

def get_user_token(headers):
    return getattr(headers, "x_forwarded_access_token", "")

def connect_with_obo(http_path: str, user_token: str):
    global _connection
    if _connection:
        return _connection
    _connection = sql.connect(
        server_hostname=cfg.host,
        http_path=http_path,
        access_token=user_token
    )
    return _connection

def execute_query(table_name: str, conn):
    with conn.cursor() as cursor:
        query = f"SELECT * FROM {table_name} LIMIT 10"
        cursor.execute(query)
        return cursor.fetchall_arrow().to_pandas()

class OboState(rx.State):
    warehouse_paths: dict[str, str] = {}
    selected_warehouse: str = ""
    table_name: str = "samples.nyctaxi.trips"
    data: list[list] = []
    columns: list[dict] = []
    is_loading: bool = False
    error_message: str = ""

    @rx.event(background=True)
    async def run_query(self):
        async with self:
            self.is_loading = True
            self.error_message = ""
            headers = self.router.headers
            user_token = get_user_token(headers)

        if not user_token:
            async with self:
                self.error_message = "No OBO token found."
                self.is_loading = False
            return

        try:
            http_path = self.warehouse_paths.get(self.selected_warehouse)
            conn = connect_with_obo(http_path, user_token)
            df = execute_query(self.table_name, conn)

            # Process dataframe for display
            d = df.values.tolist()
            c = [{"title": col, "id": col, "type": "str"} for col in df.columns]

            async with self:
                self.data = d
                self.columns = c
        except Exception as e:
            async with self:
                self.error_message = str(e)
        finally:
            async with self:
                self.is_loading = False
```

:::info

This sample caches the SQL connection in a module-level `_connection` variable so it can be reused within the same app process. The app will only work when deployed to Databricks Apps with on-behalf-of-user authentication enabled.

:::

:::warning

You need to enable [on-behalf-of-user authentication](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/app-development#-using-the-databricks-apps-authorization-model) for your application for this sample to work. When running this code locally, the `X-Forwarded-Access-Token` will not be present and the sample will not work as intended.

:::

## Resources

- [SQL warehouse](https://docs.databricks.com/aws/en/compute/sql-warehouse/)
- [Unity Catalog table](https://docs.databricks.com/aws/en/tables/)

## Permissions

For the on-behalf-of-user authentication model, permissions work as follows:

- **User's permissions**: When using OBO authentication, the query runs with the end user's permissions
  - User needs `SELECT` permissions on the tables being queried
  - User needs `CAN USE` on the SQL warehouse

- **App service principal**: When falling back to service principal authentication
  - Needs `CAN USE` on the SQL warehouse
  - Needs `SELECT` on the Unity Catalog tables for fallback access

See [Databricks Apps authorization model](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) for more information.

## Dependencies

- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Databricks SQL Connector](https://pypi.org/project/databricks-sql-connector/) - `databricks-sql-connector`
- [Pandas](https://pypi.org/project/pandas/) - `pandas`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`

```python title="requirements.txt"
databricks-sdk
databricks-sql-connector
pandas
reflex
```

---

## Embed a dashboard(Bi)


This recipe embeds a [Databricks AI/BI dashboard](https://docs.databricks.com/aws/en/dashboards/) into a Databricks App.

## Code snippet

```python title="app.py"

from databricks.sdk.core import Config

# 1. Helper to fetch published dashboards from Databricks API
def get_published_dashboards():
    cfg = Config()
    headers = {"Authorization": f"Bearer {cfg.token}"}
    # The Lakeview API endpoint for listing dashboards
    url = f"{cfg.host}/api/2.0/lakeview/dashboards"

    response = requests.get(url, headers=headers)
    response.raise_for_status()

    data = response.json()
    published = {}
    for d in data.get("dashboards", []):
        if d.get("published"):
            published[d.get("display_name")] = d.get("dashboard_id")
    return published

class AiBiDashboardState(rx.State):
    dashboard_options: dict[str, str] = {}
    selected_dashboard: str = ""
    iframe_source: str = ""

    @rx.var
    def dashboard_names(self) -> list[str]:
        return list(self.dashboard_options.keys())

    @rx.event
    def on_load(self):
        # Fetch available dashboards on load
        self.dashboard_options = get_published_dashboards()
        if self.dashboard_options:
            # Default to the first one
            self.selected_dashboard = self.dashboard_names[0]
            self.update_src()

    @rx.event
    def set_selected_dashboard(self, value: str):
        self.selected_dashboard = value
        self.update_src()

    def update_src(self):
        # Construct the embed URL
        cfg = Config()
        dash_id = self.dashboard_options.get(self.selected_dashboard)
        if dash_id:
            self.iframe_source = f"{cfg.host}/dashboardsv3/{dash_id}/published?embed=true"
```

:::info

Copy and paste the dashboard embedding URL from the dashboard UI **Share** -> **Embed iframe**.

:::

## Resources

- [AI/BI dashboard](https://docs.databricks.com/aws/en/dashboards/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN VIEW` permission on the dashboard

:::info

A workspace admin needs to enable dashboard embedding in the Security settings of your Databricks workspace for specific domains (e.g., databricksapps.com) or all domains for this sample to work.

:::

## Dependencies

- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Requests](https://pypi.org/project/requests/) - `requests`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`

```python title="requirements.txt"
databricks-sdk
requests
reflex
```

---

## Chat with a Genie Space(Bi)


This app uses the [AI/BI Genie](https://www.databricks.com/product/ai-bi) [Conversations API](https://docs.databricks.com/api/workspace/genie) to let users ask questions about your data for instant insights.

## Code snippet

For a complete working Reflex example (UI + state management), see the implementation in this repository under `reflex/app/pages/genie.py` and `reflex/app/states/genie_state.py`.

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

genie_space_id = "01f0023d28a71e599b5a62f4117916d4"

def display_message(message):
    if "content" in message:
        rx.markdown(message["content"])
    if "data" in message:
        rx.data_table(message["data"])
    if "code" in message:
        rx.accordion.root(
            rx.accordion.item(
                rx.accordion.header(
                    rx.accordion.trigger("Show generated code")
                ),
                rx.accordion.content(
                    rx.code_block(message["code"], language="sql")
                ),
                value="item-1",
            ),
            type="single",
            collapsible=True,
        )

def get_query_result(statement_id):
    # For simplicity, let's say data fits in one chunk, query.manifest.total_chunk_count = 1
    result = w.statement_execution.get_statement(statement_id)
    return pd.DataFrame(
        result.result.data_array, columns=[i.name for i in result.manifest.schema.columns]
    )

def process_genie_response(response):
    for i in response.attachments:
        if i.text:
            message = {"role": "assistant", "content": i.text.content}
            display_message(message)
        elif i.query:
            data = get_query_result(response.query_result.statement_id)
            message = {
                "role": "assistant", "content": i.query.description, "data": data, "code": i.query.query
            }
            display_message(message)

class GenieState(rx.State):
    conversation_id: str = ""
    prompt: str = ""

    @rx.event
    async def send_message(self):
        if self.prompt:
            user_prompt = self.prompt
            self.prompt = ""
            # Display user message
            # Then process with assistant
            if self.conversation_id:
                conversation = w.genie.create_message_and_wait(
                    genie_space_id, self.conversation_id, user_prompt
                )
                process_genie_response(conversation)
            else:
                conversation = w.genie.start_conversation_and_wait(genie_space_id, user_prompt)
                self.conversation_id = conversation.conversation_id
                process_genie_response(conversation)
```

:::info

Copy and paste the Genie space ID from the Genie UI URL as rooms/SPACE-ID?o=.

:::

## Resources

- [Genie](https://www.databricks.com/what-aibi-genie)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `SELECT` on the Unity Catalog table
- `CAN USE` the SQL warehouse
- `CAN VIEW` the Genie Space

## Dependencies

- [Reflex](https://pypi.org/project/reflex/) - `reflex`
- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Pandas](https://pypi.org/project/pandas/) - `pandas`

```python title="requirements.txt"
reflex
databricks-sdk
pandas
```

---

## Connect to a cluster(Compute)


This recipe uses [Databricks Connect](https://docs.databricks.com/en/dev-tools/databricks-connect/python/index.html) to execute pre-defined Python or SQL code on a **shared** cluster with UI inputs.

## Code snippet

```python title="app.py"

from typing import Any
from databricks.connect import DatabricksSession

def run_spark_workload(host: str, cluster_id: str):
    spark = DatabricksSession.builder.remote(
        host=host,
        cluster_id=cluster_id
    ).getOrCreate()

    session_info = {
        "App Name": spark.conf.get("spark.app.name", "Unknown"),
        "Master URL": spark.conf.get("spark.master", "Unknown"),
    }

    query = "SELECT 'I''m a stellar cook!' AS message"
    df_sql = spark.sql(query).toPandas()

    df_range = spark.range(10).toPandas()

    return session_info, df_sql.values.tolist(), df_range.values.tolist()

class ConnectClusterState(rx.State):
    cluster_id: str = ""
    session_info: dict = {}
    sql_output: list[list[Any]] = []
    range_output: list[list[Any]] = []
    is_loading: bool = False
    error_message: str = ""
    success_message: str = ""

    @rx.event(background=True)
    async def connect_and_run(self):
        async with self:
            self.is_loading = True
            self.error_message = ""
            self.success_message = ""
        try:
            host = os.getenv("DATABRICKS_HOST")
            loop = asyncio.get_running_loop()
            info, sql_res, range_res = await loop.run_in_executor(
                None, run_spark_workload, host, self.cluster_id
            )
            async with self:
                self.session_info = info
                self.sql_output = sql_res
                self.range_output = range_res
                self.success_message = "Successfully connected to Spark"
        except Exception as e:
            async with self:
                self.error_message = str(e)
        finally:
            async with self:
                self.is_loading = False
```

:::info

You also have the option to [connect to serverless compute using Databricks Connect](https://docs.databricks.com/aws/en/compute/serverless/).

:::

## Resources

- [All-purpose compute](https://docs.databricks.com/aws/en/compute/use-compute) or [serverless compute](https://docs.databricks.com/aws/en/compute/serverless/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN ATTACH TO` permission on the cluster

See [Compute permissions](https://docs.databricks.com/aws/en/compute/clusters-manage#compute-permissions) for more information.

## Dependencies

- [Databricks Connect](https://pypi.org/project/databricks-connect/) - `databricks-connect`
- [Pandas](https://pypi.org/project/pandas/) - `pandas`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`

```python title="requirements.txt"
databricks-connect
pandas
reflex
```

---

## External connections(External_services)


This recipe demonstrates how to use Unity Catalog-managed external [HTTP connections](https://docs.databricks.com/aws/en/query-federation/http) for secure and governed access to MCP and non-MCP servers, for example, to GitHub, or Jira, and Slack.

## Code snippets

### OAuth User to Machine Per User (On-behalf-of-user)

```python title="app.py"

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ExternalFunctionRequestHttpMethod

def init_mcp_session(w: WorkspaceClient, connection_name: str):
    init_payload = {
        "jsonrpc": "2.0",
        "id": "init-1",
        "method": "initialize",
        "params": {
            "protocolVersion": "2024-11-05",
            "capabilities": {},
            "clientInfo": {"name": "reflex-cookbook", "version": "1.0"},
        }
    }
    response = w.serving_endpoints.http_request(
        conn=connection_name,
        method=ExternalFunctionRequestHttpMethod.POST,
        path="/",
        json=init_payload,
    )
    return response.headers.get("mcp-session-id")

class ExternalConnectionsState(rx.State):
    response_data: str = ""

    @rx.event
    async def run(self):
        # 1. Get token from headers
        headers = self.router.headers
        token = getattr(headers, "x_forwarded_access_token", "")

        # 2. Initialize WorkspaceClient with the token
        w = WorkspaceClient(token=token, auth_type="pat")

        # 3. Initialize Session
        connection_name = "github_mcp_oauth"
        session_id = init_mcp_session(w, connection_name)

        # 4. Make request with session ID
        headers = {"Mcp-Session-Id": session_id}
        payload = {"jsonrpc": "2.0", "id": "list-1", "method": "tools/list"}

        response = w.serving_endpoints.http_request(
            conn=connection_name,
            method=ExternalFunctionRequestHttpMethod.POST,
            path="/",
            headers=headers,
            json=payload,
        )
        self.response_data = json.dumps(response.as_dict(), indent=2)
```

### Bearer token

```python title="app.py"

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ExternalFunctionRequestHttpMethod

class ExternalConnectionsState(rx.State):
    response_data: str = ""

    @rx.event
    async def run(self):
        # 1. Initialize WorkspaceClient (uses environment or default auth)
        w = WorkspaceClient()

        # 2. Direct HTTP GET request
        response = w.serving_endpoints.http_request(
            conn="github_u2m_connection",
            method=ExternalFunctionRequestHttpMethod.GET,
            path="/",
            headers={"Accept": "application/vnd.github+json"},
        )

        self.response_data = json.dumps(response.as_dict(), indent=2)
```

## Resources

- [Unity Catalog HTTP Connection](https://docs.databricks.com/aws/en/query-federation/http), either MCP or non-MCP

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CONNECTION` permission on the HTTP Connection

When using OAuth User to Machine Per User (On-behalf-of-user), you need to configure [User authorization](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/auth#user-authorization) by adding the Unity Catalog connection or other scopes.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`

```python title="requirements.txt"
databricks-sdk
reflex
```

---

## Connect an OLTP database


This app connects to a [Databricks Lakebase](https://docs.databricks.com/aws/en/oltp/) OLTP database instance for reads and writes, e.g., of an App state. Provide the instance name, database, schema, and state table.

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

from psycopg_pool import ConnectionPool
from typing import Union, Optional

# This module-level variable will cache the connection pool across the app.
_pool: Optional[ConnectionPool] = None

# Define a custom connection class to handle token rotation.
class RotatingTokenConnection(psycopg.Connection):
    def __init__(self, *args, **kwargs):
        self._instance_name = kwargs.pop("_instance_name")
        super().__init__(*args, **kwargs)

    @classmethod
    def connect(cls, *args, **kwargs):
        if "_instance_name" in kwargs:
            instance_name = kwargs["_instance_name"]
            # Generate a fresh OAuth token for each new connection.
            w = WorkspaceClient()
            credential = w.database.generate_database_credential(
                request_id=str(uuid.uuid4()), instance_names=[instance_name]
            )
            kwargs["password"] = credential.token
        return super().connect(*args, **kwargs)

def build_pool(instance_name: str, host: str, user: str, database: str) -> ConnectionPool:
    """Builds and returns a new connection pool with token rotation."""
    # Note: sslmode is set to require to ensure encrypted connections.
    return ConnectionPool(
        conninfo=f"host={host} dbname={database} user={user} sslmode=require",
        min_size=1,
        max_size=5,
        open=True,
        kwargs={"_instance_name": instance_name},
        connection_class=RotatingTokenConnection,
    )

def query_df(query: str, params=None) -> pd.DataFrame:
    """Executes a query using the global pool and returns a DataFrame."""
    global _pool
    if _pool is None:
        raise ConnectionError("Connection pool is not initialized.")
    with _pool.connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(query, params)
            if cursor.description is None:
                return pd.DataFrame()
            cols = [desc[0] for desc in cursor.description]
            return pd.DataFrame(cursor.fetchall(), columns=cols)

def upsert_app_state(schema: str, table: str, session_id: str, key: str, value: str):
    """Helper to create a table and upsert a key-value pair for a session."""
    create_sql = f"""
    CREATE TABLE IF NOT EXISTS {schema}.{table} (
        session_id TEXT,
        key TEXT,
        value TEXT,
        updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
        PRIMARY KEY (session_id, key)
    );
    """
    query_df(create_sql)

    upsert_sql = f"""
    INSERT INTO {schema}.{table} (session_id, key, value, updated_at) 
    VALUES (%(session_id)s, %(key)s, %(value)s, CURRENT_TIMESTAMP)
    ON CONFLICT (session_id, key) DO UPDATE SET
        value = EXCLUDED.value,
        updated_at = EXCLUDED.updated_at;
    """
    query_df(upsert_sql, params={"session_id": session_id, "key": key, "value": value})

class OltpDatabaseState(rx.State):
    session_id: str = str(uuid.uuid4())
    selected_instance: str = "your_instance_name"
    database: str = "databricks_postgres"
    schema_name: str = "public"
    table_name: str = "app_state"
    result_data: list[dict[str, Union[str, int, float, bool, None, datetime.datetime]]] = []
    is_loading: bool = False
    error_message: str = ""

    @rx.event(background=True)
    async def run_query(self):
        global _pool
        async with self:
            self.is_loading = True
            self.error_message = ""
        try:
            # Initialize the pool if it's the first run.
            if _pool is None:
                w = WorkspaceClient()
                user = w.current_user.me().user_name
                instance = w.database.get_database_instance(name=self.selected_instance)
                host = instance.read_write_dns
                _pool = build_pool(
                    instance_name=self.selected_instance,
                    host=host, user=user, database=self.database
                )

            # Create table and insert/update a record for this session.
            upsert_app_state(
                self.schema_name, self.table_name, self.session_id, "feedback_message", "true"
            )

            # Fetch the data to display.
            select_sql = f"SELECT * FROM {self.schema_name}.{self.table_name} WHERE session_id = %(session_id)s"
            df = query_df(select_sql, params={"session_id": self.session_id})

            async with self:
                self.result_data = df.to_dict("records")

        except Exception as e:
            async with self:
                self.error_message = f"An error occurred: {e}"
        finally:
            async with self:
                self.is_loading = False
```

:::info

This sample keeps a module-level `ConnectionPool` in `_pool` so connections can be reused within the same app process. For multi-worker deployments, configure pooling appropriately for your workload.

:::

## Resources

- [Lakebase](https://docs.databricks.com/aws/en/oltp/) database instance (PostgreSQL).
- Target PostgreSQL database/schema/table.

## Permissions

First, the database instance should be specified in your [**App resources**](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/resources).

Then, your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

```sql
GRANT CONNECT ON DATABASE databricks_postgres TO "099f0306-9e29-4a87-84c0-3046e4bcea02";
GRANT USAGE, CREATE ON SCHEMA public TO "099f0306-9e29-4a87-84c0-3046e4bcea02";
GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE app_state TO "099f0306-9e29-4a87-84c0-3046e4bcea02";
```

See [this guide](https://docs.databricks.com/aws/en/oltp/pg-roles?language=PostgreSQL#create-postgres-roles-and-grant-privileges-for-databricks-identities) for more information.

[This guide](https://learn.microsoft.com/en-us/azure/databricks/oltp/query/sql-editor#create-a-new-query) shows you how to query your Lakebase.

## Dependencies

- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk>=0.60.0`
- [`psycopg[binary]`](https://pypi.org/project/psycopg/), [`psycopg-pool`](https://pypi.org/project/psycopg-pool/)
- [Pandas](https://pypi.org/project/pandas/) - `pandas`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`

```python title="requirements.txt"
databricks-sdk
pandas
reflex
psycopg[binary]
psycopg-pool
```

---

## Edit a Delta table(Tables)


Use this recipe to read, edit, and write back data in a [Unity Catalog table](https://docs.databricks.com/aws/en/tables/) using the [Databricks SQL Connector](https://docs.databricks.com/en/dev-tools/python-sql-connector.html).

## Code snippet

```python title="app.py"

from typing import Any

from databricks import sql
from databricks.sdk.core import Config

_connection = None

def get_connection(http_path: str):
    global _connection
    if _connection:
        return _connection
    cfg = Config()
    connection = sql.connect(
        server_hostname=cfg.host,
        http_path=http_path,
        credentials_provider=cfg.authenticate,
    )
    _connection = connection
    return connection

def insert_overwrite_table(table_name: str, df: pd.DataFrame, connection):
    if df.empty:
        # If the dataframe is empty, we skip the overwrite to avoid
        # accidentally truncating the table if the UI state was not fully loaded.
        return

    def format_val(x):
        if x is None or pd.isna(x):
            return "NULL"
        if isinstance(x, str):
            return "'" + x.replace("'", "''") + "'"
        if isinstance(x, (datetime.date, datetime.datetime, pd.Timestamp)):
            return f"'{x}'"
        return str(x)

    values = []
    for _, row in df.iterrows():
        row_values = [format_val(val) for val in row]
        values.append(f"({', '.join(row_values)})")

    sql_query = f"INSERT OVERWRITE TABLE {table_name} VALUES {', '.join(values)}"

    with connection.cursor() as cursor:
        cursor.execute(sql_query)

class EditDeltaTableState(rx.State):
    # Key state fields for managing selection and data
    warehouse_paths: dict[str, str] = {}
    selected_warehouse: str = ""
    selected_catalog: str = ""
    selected_schema: str = ""
    selected_table: str = ""

    # Data for the editor
    columns: list[dict[str, str]] = []
    table_data: list[list[Any]] = []
    original_table_data: list[list[Any]] = []

    is_saving: bool = False

    @rx.event
    def handle_cell_change(
        self, new_value: Any, row_index: int, col_index: int
    ):
        "Update table data when a cell is edited."
        if row_index < len(self.table_data):
            row = self.table_data[row_index]
            if col_index < len(row):
                self.table_data[row_index][col_index] = new_value

    @rx.event
    async def save_changes(self):
        self.is_saving = True
        yield
        try:
            col_names = [col["title"] for col in self.columns]
            df = pd.DataFrame(self.table_data, columns=col_names)

            http_path = self.warehouse_paths.get(self.selected_warehouse)
            full_table_name = f"{self.selected_catalog}.{self.selected_schema}.{self.selected_table}"
            conn = get_connection(http_path)

            insert_overwrite_table(full_table_name, df, conn)

            self.original_table_data = [row[:] for row in self.table_data]
            yield rx.toast(f"Successfully saved changes to {full_table_name}.")
        except Exception as e:
            yield rx.toast(f"Error saving changes: {e}", level="error")
        finally:
            self.is_saving = False
```

:::info

This sample caches the SQL connection in a module-level `_connection` variable so it can be reused within the same app process. For multi-worker deployments, consider a proper pooling strategy or per-request connections depending on your workload.

:::

## Resources

- [SQL warehouse](https://docs.databricks.com/aws/en/compute/sql-warehouse/)
- [Unity Catalog table](https://docs.databricks.com/aws/en/tables/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `MODIFY` on the Unity Catalog table
- `CAN USE` on the SQL warehouse

See Unity [Catalog privileges and securable objects](https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges) for more information.

## Dependencies

- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Databricks SQL Connector](https://pypi.org/project/databricks-sql-connector/) - `databricks-sql-connector`
- [Pandas](https://pypi.org/project/pandas/) - `pandas`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`

```python title="requirements.txt"
databricks-sdk
databricks-sql-connector
pandas
reflex
```

---

## Read a Delta table(Tables)


This recipe reads a [Unity Catalog table](https://docs.databricks.com/aws/en/tables/) using the [Databricks SQL Connector](https://docs.databricks.com/en/dev-tools/python-sql-connector.html).

## Code snippet

```python title="app.py"

from databricks import sql
from databricks.sdk.core import Config

from typing import Any

_connection = None

def get_connection(http_path: str):
    global _connection
    if _connection:
        return _connection
    cfg = Config()
    connection = sql.connect(
        server_hostname=cfg.host,
        http_path=http_path,
        credentials_provider=cfg.authenticate,
    )
    _connection = connection
    return connection

def read_table(table_name: str, conn) -> pd.DataFrame:
    with conn.cursor() as cursor:
        cursor.execute(f"SELECT * FROM {table_name}")
        return cursor.fetchall_arrow().to_pandas()

def pandas_to_editor_format(
    df: pd.DataFrame,
) -> tuple[list[list[Any]], list[dict[str, str]]]:
    """Convert a pandas DataFrame to the format required by rx.data_editor."""
    if df.empty:
        return ([], [])
    data = df.values.tolist()
    columns = [{"title": col, "id": col, "type": "str"} for col in df.columns]
    return (data, columns)

class ReadTableState(rx.State):
    http_path_input: str = ""
    table_name: str = "samples.nyctaxi.trips"
    df_data: list[list[Any]] = []
    df_columns: list[dict] = []
    is_loading: bool = False
    error_message: str = ""

    @rx.var
    def columns_for_editor(self) -> list[dict]:
        return self.df_columns

    @rx.var
    def data_for_editor(self) -> list[list[str]]:
        """Return data directly as it is already formatted for the editor."""
        return self.df_data

    @rx.event(background=True)
    async def load_table(self):
        async with self:
            self.is_loading = True
            self.error_message = ""
            self.df_data = []
            self.df_columns = []
            http_path = self.http_path_input
            table_name = self.table_name
        if not http_path:
            async with self:
                self.error_message = "Please enter an HTTP Path."
                self.is_loading = False
            return
        try:
            conn = get_connection(http_path)
            df = read_table(table_name, conn)
            data, cols = pandas_to_editor_format(df)
            async with self:
                self.df_data = data
                self.df_columns = cols
        except Exception as e:
            logging.exception(f"Error loading Delta table: {e}")
            async with self:
                self.error_message = f"Error: {e}"
        finally:
            async with self:
                self.is_loading = False
```

:::info

This sample caches the SQL connection in a module-level `_connection` variable so it can be reused within the same app process. For multi-worker deployments, consider a proper pooling strategy or per-request connections depending on your workload.

:::

## Resources

- [SQL warehouse](https://docs.databricks.com/aws/en/compute/sql-warehouse/)
- [Unity Catalog table](https://docs.databricks.com/aws/en/tables/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `SELECT` on the Unity Catalog table
- `CAN USE` on the SQL warehouse

See Unity [Catalog privileges and securable objects](https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges) for more information.

## Dependencies

- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Databricks SQL Connector](https://pypi.org/project/databricks-sql-connector/) - `databricks-sql-connector`
- [Pandas](https://pypi.org/project/pandas/) - `pandas`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`

```python title="requirements.txt"
databricks-sdk
databricks-sql-connector
pandas
reflex
```

---

## Download a file(Volumes)


This recipe downloads a file from a [Unity Catalog volume](https://docs.databricks.com/en/volumes/index.html) using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/).

:::note

Unlike notebooks, Databricks Apps does not support mounting Unity Catalog volumes and directly reading and writing files. As this code snippet demonstrates, each file needs to be downloaded to the app compute before being able to manipulate it.

:::

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

class DownloadFileState(rx.State):
    download_file_path: str = ""

    async def handle_download(self):
        response = w.files.download(self.download_file_path)
        file_data = response.contents.read()
        file_name = os.path.basename(self.download_file_path)

        return rx.download(data=file_data, filename=file_name)
```

## Resources

- [Unity Catalog volume](https://docs.databricks.com/aws/en/files/volumes)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CATALOG` on the volume's catalog
- `USE SCHEMA` on the volume's schema
- `READ VOLUME` on the volume

See [Privileges required for volume operations](https://docs.databricks.com/en/volumes/privileges.html#privileges-required-for-volume-operations) for more information.

If you declare volume access in a Databricks Asset Bundle, `resources.apps[*].resources[*].uc_securable` may not grant `USE_CATALOG` and `USE_SCHEMA` on the parent catalog and schema (the app still needs them at runtime). As a temporary workaround until bundles can declare those parent grants, add the privileges manually, or see [apps_grants_sync](https://github.com/salihbout/apps_grants_sync): an example Databricks App and Asset Bundle that wires `experimental.scripts.postdeploy` so parent privileges are applied after each `databricks bundle deploy` (copy its `tools/` into your bundle or mirror the same pattern in `databricks.yml`).

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`

```python title="requirements.txt"
databricks-sdk
reflex
```

---

## Upload a file(Volumes)


This recipe uploads a file to a [Unity Catalog volume](https://docs.databricks.com/en/volumes/index.html) using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/).

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

def check_upload_permissions(volume_name: str):
    try:
        volume = w.volumes.read(name=volume_name)
        current_user = w.current_user.me()
        grants = w.grants.get_effective(
            securable_type="volume",
            full_name=volume.full_name,
            principal=current_user.user_name,
        )

        if not grants or not grants.privilege_assignments:
            return "Insufficient permissions: No grants found."

        for assignment in grants.privilege_assignments:
            for privilege in assignment.privileges:
                if privilege.privilege.value in ["ALL_PRIVILEGES", "WRITE_VOLUME"]:
                    return "Volume and permissions validated"

        return "Insufficient permissions: Required privileges not found."
    except Exception as e:
        return f"Error: {e}"

class UploadFileState(rx.State):
    upload_volume_path: str = ""
    volume_check_success: bool = False
    permission_result: str = ""
    is_checking: bool = False
    is_uploading: bool = False
    uploaded_files: list[str] = []

    @rx.event
    async def check_volume_permissions(self):
        self.is_checking = True
        self.permission_result = ""
        self.volume_check_success = False
        yield

        result = check_upload_permissions(self.upload_volume_path)

        self.permission_result = result
        if "validated" in result:
            self.volume_check_success = True
        self.is_checking = False

    @rx.event
    async def handle_upload(self, files: list[rx.UploadFile]):
        if not files: return
        self.is_uploading = True
        yield
        try:
            parts = self.upload_volume_path.strip().split(".")
            catalog, schema, volume_name = parts[0], parts[1], parts[2]
            for file in files:
                file_bytes = await file.read()
                binary_data = io.BytesIO(file_bytes)
                path = f"/Volumes/{catalog}/{schema}/{volume_name}/{file.filename}"
                w.files.upload(path, binary_data, overwrite=True)
                self.uploaded_files.append(file.filename)
                yield rx.toast(f"Uploaded {file.filename} to {path}", level="success")
        except Exception as e:
            yield rx.toast(f"Upload failed: {e}", level="error")
        finally:
            self.is_uploading = False
```

## Resources

- [Unity Catalog volume](https://docs.databricks.com/aws/en/files/volumes)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CATALOG` on the catalog of the volume
- `USE SCHEMA` on the schema of the volume
- `READ VOLUME` and `WRITE VOLUME` on the volume

See [Privileges required for volume operations](https://docs.databricks.com/en/volumes/privileges.html#privileges-required-for-volume-operations) for more information.

If you declare volume access in a Databricks Asset Bundle, `resources.apps[*].resources[*].uc_securable` may not grant `USE_CATALOG` and `USE_SCHEMA` on the parent catalog and schema (the app still needs them at runtime). As a temporary workaround until bundles can declare those parent grants, add the privileges manually, or see [apps_grants_sync](https://github.com/salihbout/apps_grants_sync): an example Databricks App and Asset Bundle that wires `experimental.scripts.postdeploy` so parent privileges are applied after each `databricks bundle deploy` (copy its `tools/` into your bundle or mirror the same pattern in `databricks.yml`).

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`

```python title="requirements.txt"
databricks-sdk
reflex
```

---

## Retrieve workflow results(Workflows)


This recipe retrieves the results of a [Databricks Workflows](https://docs.databricks.com/en/jobs/index.html) job run using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/).

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

class RetrieveJobResultsState(rx.State):
    run_id: str = ""
    results: str = ""
    is_loading: bool = False
    error_message: str = ""

    @rx.event(background=True)
    async def get_results(self):
        async with self:
            self.is_loading = True
            self.error_message = ""
            self.results = ""

            if not self.run_id:
                yield rx.toast("Please specify a Run ID.", level="warning")
                self.is_loading = False
                return

        try:
            # Ensure ID is an integer
            run_id = int(self.run_id)
            run_output = w.jobs.get_run_output(run_id=run_id)

            async with self:
                self.results = str(run_output)
                yield rx.toast("Results retrieved successfully.", level="success")
        except ValueError:
            async with self:
                self.error_message = "Run ID must be a valid number."
                yield rx.toast("Run ID must be a valid number.", level="error")
        except Exception as e:
            async with self:
                self.error_message = str(e)
                yield rx.toast(f"Error: {e}", level="error")
        finally:
            async with self:
                self.is_loading = False
```

## Resources

- [Job](https://docs.databricks.com/aws/en/jobs/configure-job)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN VIEW` permission on the job

See [Control access to a job](https://docs.databricks.com/en/jobs/privileges.html#control-access-to-a-job) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`

```python title="requirements.txt"
databricks-sdk
reflex
```

---

## Run a workflow(Workflows)


This recipe triggers a [Databricks Workflows](https://docs.databricks.com/en/jobs/index.html) job using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/).

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

def trigger_workflow(job_id: str, params: dict):
    w = WorkspaceClient()
    try:
        run = w.jobs.run_now(job_id=int(job_id), job_parameters=params)
        return {"run_id": run.run_id, "number_in_job": run.number_in_job}
    except Exception as e:
        return {"error": str(e)}

class TriggerJobState(rx.State):
    job_id: str = ""
    parameters_input: str = ""
    result_data: str = ""
    error_message: str = ""
    is_loading: bool = False

    @rx.event(background=True)
    async def trigger_job(self):
        async with self:
            self.is_loading = True
            self.error_message = ""
            self.result_data = ""

            if not self.job_id:
                yield rx.toast("Please specify a Job ID.", level="warning")
                self.is_loading = False
                return

        try:
            params = json.loads(self.parameters_input.strip())
            result = trigger_workflow(self.job_id, params)

            async with self:
                if "error" in result:
                    self.error_message = result["error"]
                    yield rx.toast(f"Error: {result['error']}", level="error")
                else:
                    self.result_data = json.dumps(result, indent=2)
                    yield rx.toast(f"Run started: {result.get('run_id')}", level="success")
        except json.JSONDecodeError:
            async with self:
                yield rx.toast("Invalid JSON parameters", level="error")
        except Exception as e:
            async with self:
                self.error_message = str(e)
        finally:
            async with self:
                self.is_loading = False
```

## Resources

- [Job](https://docs.databricks.com/aws/en/jobs/configure-job)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN MANAGE RUN` permission on the job

See [Control access to a job](https://docs.databricks.com/en/jobs/privileges.html#control-access-to-a-job) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Reflex](https://pypi.org/project/reflex/) - `reflex`

```python title="requirements.txt"
databricks-sdk
reflex
```

---

## Connect an MCP server(3)


This recipe connects to an [MCP](https://modelcontextprotocol.io/overview) server for AI applications using GitHub as an example and Unity Catalog [HTTP connections](https://docs.databricks.com/aws/en/query-federation/http) for secure and governed access.

## Code snippets

### OAuth User to Machine Per User (On-behalf-of-user)

```python title="app.py"

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ExternalFunctionRequestHttpMethod

token = st.context.headers.get("x-forwarded-access-token")
w = WorkspaceClient(token=token, auth_type="pat")

def init_mcp_session(w: WorkspaceClient, connection_name: str):
    init_payload = {
        "jsonrpc": "2.0",
        "id": "init-1",
        "method": "initialize",
        "params": {}
    }
    response = w.serving_endpoints.http_request(
        conn=connection_name,
        method=ExternalFunctionRequestHttpMethod.POST,
        path="/",
        json=init_payload,
    )
    return response.headers.get("mcp-session-id")

connection_name = "github_u2m_connection"
http_method = ExternalFunctionRequestHttpMethod.POST
path = "/"
headers = {"Content-Type": "application/json"}
payload = {"jsonrpc": "2.0", "id": "list-1", "method": "tools/list"}

if st.button("Run"):
    session_id = init_mcp_session(w, connection_name)
    headers["Mcp-Session-Id"] = session_id

    response = w.serving_endpoints.http_request(
        conn=connection_name,
        method=http_method,
        path=path,
        headers=headers,
        json=payload,
    )
    st.json(response.json())
```

### Bearer token

```python title="app.py"

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ExternalFunctionRequestHttpMethod

w = WorkspaceClient()

response = w.serving_endpoints.http_request(
    conn="github_u2m_connection",
    method=ExternalFunctionRequestHttpMethod.GET,
    path="/traffic/views",
    headers={"Accept": "application/vnd.github+json"},
    json={
        "jsonrpc": "2.0",
        "id": "init-1",
        "method": "initialize",
        "params": {}
    },
)

st.json(response.json())
```

## Resources

- [Unity Catalog HTTP Connection](https://docs.databricks.com/aws/en/query-federation/http) with the MCP (/mcp) base path

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CONNECTION` permission on the HTTP Connection

When using OAuth User to Machine Per User (On-behalf-of-user), you need to configure [User authorization](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/auth#user-authorization) by adding the Unity Catalog connection or other scopes.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`
- [MCP CLI](https://pypi.org/project/mcp/) - `mcp[cli]`

```python title="requirements.txt"
databricks-sdk
streamlit
mcp[cli]
```

---

## Invoke a model(3)


This recipe invokes a model hosted on [Mosaic AI Model Serving](https://docs.databricks.com/aws/en/machine-learning/model-serving/) using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/) and returns the result. Choose either a traditional ML model or a large language model (LLM).

## Code snippets

### Traditional Machine Learning

#### Using `dataframe_split` (JSON-serialized DataFrame in split orientation)

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="custom-regression-model",
    dataframe_split={
        "columns": ["feature1", "feature2"],
        "data": [[1.5, 2.5]]
    }
)
st.json(response.as_dict())
```

#### Using `dataframe_records` (JSON-serialized DataFrame in records orientation)

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="custom-regression-model",
    dataframe_records={
        "feature1": [1.5],
        "feature2": [2.5]
    }
)
st.json(response.as_dict())
```

#### Using `instances` (Tensor inputs in row format for TensorFlow/PyTorch models)

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

tensor_input = [[1.0, 2.0, 3.0]]
response = w.serving_endpoints.query(
    name="tensor-processing-model",
    instances=tensor_input,
)
st.json(response.as_dict())
```

#### Using `inputs` (Tensor inputs in columnar format for TensorFlow/PyTorch models)

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

tensor_input = {
    "input1": [1.0, 2.0, 3.0],
    "input2": [4.0, 5.0, 6.0],
}
response = w.serving_endpoints.query(
    name="tensor-processing-model",
    inputs=tensor_input,
)
st.json(response.as_dict())
```

### Large language models (LLMs)

#### Using `prompt` (Input text for completion tasks)

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="llm-text-completions-model",
    prompt="Generate a recipe for building scalable Databricks Apps.",
    temperature=0.5,
)
st.json(response.as_dict())
```

#### Using `messages` (List of chat messages for conversational models)

```python title="app.py"

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ChatMessage, ChatMessageRole

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="chat-assistant-model",
    messages=[
        ChatMessage(
            role=ChatMessageRole.SYSTEM,
            content="You are a helpful assistant.",
        ),
        ChatMessage(
            role=ChatMessageRole.USER,
            content="Provide tips for deploying Databricks Apps.",
        ),
    ],
)
st.json(response.as_dict())
```

#### Using `input` (Input text for embedding tasks)

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="embedding-model",
    input="Databricks provides a unified analytics platform.",
)
st.json(response.as_dict())
```

## Resources

- [Model Serving endpoint](https://docs.databricks.com/aws/en/machine-learning/model-serving/manage-serving-endpoints)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN QUERY` on the model serving endpoint

See [Manage permissions on your model serving endpoint](https://docs.databricks.com/aws/en/machine-learning/model-serving/manage-serving-endpoints#manage-permissions-on-your-model-serving-endpoint) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`

```python title="requirements.txt"
databricks-sdk
streamlit
```

---

## Run vector search(3)


This recipe performs a vector search query on a [Mosaic AI Vector Search](https://docs.databricks.com/aws/en/generative-ai/vector-search) index using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/).

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

openai_client = w.serving_endpoints.get_open_ai_client()

EMBEDDING_MODEL_ENDPOINT_NAME = "databricks-gte-large-en"

def get_embeddings(text):
    try:
        response = openai_client.embeddings.create(
            model=EMBEDDING_MODEL_ENDPOINT_NAME, input=text
        )
        return response.data[0].embedding
    except Exception as e:
        st.text(f"Error generating embeddings: {e}")

def run_vector_search(prompt: str) -> str:
    prompt_vector = get_embeddings(prompt)
    if prompt_vector is None or isinstance(prompt_vector, str):
        return f"Failed to generate embeddings: {prompt_vector}"

    columns_to_fetch = [col.strip() for col in columns.split(",") if col.strip()]

    try:
        query_result = w.vector_search_indexes.query_index(
            index_name=index_name,
            columns=columns_to_fetch,
            query_vector=prompt_vector,
            num_results=3,
        )
        return query_result.result.data_array
    except Exception as e:
        return f"Error during vector search: {e}"

index_name = st.text_input(
    label="Unity Catalog Vector search index:",
    placeholder="catalog.schema.index-name",
)

columns = st.text_input(
    label="Columns to retrieve (comma-separated):",
    placeholder="url, name",
    help="Enter one or more column names present in the vector search index, separated by commas. E.g. id, text, url.",
)

text_input = st.text_input(
    label="Enter your search query:",
    placeholder="What is Databricks?",
    key="search_query_key",
)

if st.button("Run vector search"):
    result = run_vector_search(text_input)
    st.write("Search results:")
    st.write(result)
```

## Resources

- [Vector Search endpoint](https://docs.databricks.com/aws/en/generative-ai/vector-search)
- [Vector Search index](https://docs.databricks.com/aws/en/generative-ai/vector-search)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CATALOG` on the catalog that contains the Vector Search index
- `USE SCHEMA` on the schema that contains the Vector Search index
- `SELECT` on the Vector Search index

See [Query a vector search endpoint](https://docs.databricks.com/aws/en/generative-ai/vector-search) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`

```python title="requirements.txt"
databricks-sdk
streamlit
```

---

## Retrieve a secret(3)


This recipe retrieves a [Databricks secret](https://docs.databricks.com/en/security/secrets/index.html). Use secrets to securely connect to external services and APIs.

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

def get_secret(scope, key):
    try:
        secret_response = w.secrets.get_secret(scope=scope, key=key)
        decoded_secret = base64.b64decode(secret_response.value).decode('utf-8')
        return decoded_secret
    except Exception as e:
        st.error("Secret not found or inaccessible. Please create a secret scope and key before retrieving.")

scope_name = st.text_input("Secret scope:", "my_secret_scope")
secret_key = st.text_input("Secret key (name):", "api_key")

if st.button("Retrieve"):
    secret = get_secret(scope_name, secret_key)
    st.success("Secret retrieved! The value is securely handled in the backend.")
```

## Resources

- [Secret scope and secret](https://docs.databricks.com/aws/en/security/secrets/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN READ` on the secret scope

See [Manage secret scope permissions](https://docs.databricks.com/aws/en/security/secrets/#manage-secret-scope-permissions) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`

```python title="requirements.txt"
databricks-sdk
streamlit
```

---

## Get current user(3)


This recipe gets information about the user accessing this Databricks App from their [HTTP headers](https://docs.databricks.com/en/dev-tools/databricks-apps/app-development.html#what-http-headers-are-passed-to-databricks-apps).

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

# Get information from HTTP headers
headers = st.context.headers
email = headers.get("X-Forwarded-Email")
username = headers.get("X-Forwarded-Preferred-Username")
user = headers.get("X-Forwarded-User")
ip = headers.get("X-Real-Ip")
user_access_token = headers.get("X-Forwarded-Access-Token")

# Display information from headers
st.write(f"E-mail: {email}, username: {username}, user: {user}, ip: {ip}")
st.write(f"Access token present: {'Yes' if user_access_token else 'No'}")

# If we have a user access token, we can get more information about the user
if user_access_token:
    # Initialize WorkspaceClient with the user's token
    w = WorkspaceClient(token=user_access_token, auth_type="pat")

    # Get current user information
    current_user = w.current_user.me()

    # Display user information
    st.write(f"""
        User ID: {current_user.id}
        Username: {current_user.user_name}
        Display Name: {current_user.display_name}
        Active: {current_user.active}
        Groups: {len(current_user.groups) if current_user.groups else 0} groups
        Entitlements: {len(current_user.entitlements) if current_user.entitlements else 0} entitlements
    """)
```

:::info
This sample requires on-behalf-of-user authentication to be enabled for your app to access the X-Forwarded-Access-Token header. Without this, you will only have access to basic user information from the headers, not the detailed information from the Databricks API. Without the user token present, w.current_user.me() will return information about the app service principal.
:::

## Resources

No Databricks resources are required for this recipe.

## Permissions

No permissions configuration required for accessing headers. To use the `current_user.me()` API, the app must be configured with [on-behalf-of-user authentication](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/app-development#-using-the-databricks-apps-authorization-model).

## Dependencies

- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`
- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk` (if using `current_user.me()`)

```python title="requirements.txt"
streamlit
databricks-sdk
```

---

## On-behalf-of-user authentication(Authentication)


This recipe demonstrates how to use Databricks Apps [on-behalf-of-user authentication](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/app-development#-using-the-databricks-apps-authorization-model) to run a SQL query using the user's credentials instead of the app's service principal.

## Code snippet

```python title="app.py"

from databricks import sql
from databricks.sdk.core import Config

cfg = Config()

def get_user_token():
    headers = st.context.headers
    user_token = headers["X-Forwarded-Access-Token"]
    return user_token

@st.cache_resource(ttl=300, show_spinner=True)
def connect_with_obo(http_path, user_token):
    return sql.connect(
        server_hostname=cfg.host,
        http_path=http_path,
        access_token=user_token
    )

def execute_query(table_name, conn):
    with conn.cursor() as cursor:
        query = f"SELECT * FROM {table_name} LIMIT 10"
        cursor.execute(query)
        return cursor.fetchall_arrow().to_pandas()

user_token = get_user_token()

http_path = "/sql/1.0/warehouses/abcd1234"  # Replace with your SQL warehouse HTTP path
table_name = "samples.nyctaxi.trips"  # Replace with your catalog.schema.table

if st.button("Run Query"):
    conn = connect_with_obo(http_path, user_token)
    df = execute_query(table_name, conn)
    st.dataframe(df)
```

:::info

This sample uses Streamlit's [st.cache_resource](https://docs.streamlit.io/develop/concepts/architecture/caching#stcache_resource) to cache the database connection across users, sessions, and reruns. The app will only work when deployed to Databricks Apps with on-behalf-of-user authentication enabled.

:::

:::warning

You need to enable [on-behalf-of-user authentication](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/app-development#-using-the-databricks-apps-authorization-model) for your application for this sample to work. When running this code locally, the `X-Forwarded-Access-Token` will not be present and the sample will not work as intended.

:::

## Resources

- [SQL warehouse](https://docs.databricks.com/aws/en/compute/sql-warehouse/)
- [Unity Catalog table](https://docs.databricks.com/aws/en/tables/)

## Permissions

For the on-behalf-of-user authentication model, permissions work as follows:

- **User's permissions**: When using OBO authentication, the query runs with the end user's permissions
  - User needs `SELECT` permissions on the tables being queried
  - User needs `CAN USE` on the SQL warehouse

- **App service principal**: When falling back to service principal authentication
  - Needs `CAN USE` on the SQL warehouse
  - Needs `SELECT` on the Unity Catalog tables for fallback access

See [Databricks Apps authorization model](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) for more information.

## Dependencies

- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Databricks SQL Connector](https://pypi.org/project/databricks-sql-connector/) - `databricks-sql-connector`
- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`

```python title="requirements.txt"
databricks-sdk
databricks-sql-connector
streamlit
```

---

## Embed a dashboard(3)


This recipe embeds a [Databricks AI/BI dashboard](https://docs.databricks.com/aws/en/dashboards/) into a Databricks App.

## Code snippet

```python title="app.py"

iframe_source = "https://workspace.azuredatabricks.net/embed/dashboardsv3/dashboard-id"

components.iframe(
    src=iframe_source,
    height=600,
    scrolling=True
)
```

:::info

Copy and paste the dashoard embedding URL from the dashboard UI **Share** -> **Embed iframe**.

:::

## Resources

- [AI/BI dashboard](https://docs.databricks.com/aws/en/dashboards/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN VIEW` permission on the dashboard

:::info

A workspace admin needs to enable dashboard embedding in the Security settings of your Databricks workspace for specific domains (e.g., databricksapps.com) or all domains for this sample to work.

:::

## Dependencies

- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`

```python title="requirements.txt"
streamlit
```

---

## Chat with a Genie Space(3)


This app uses the [AI/BI Genie](https://www.databricks.com/product/ai-bi) [Conversations API](https://docs.databricks.com/api/workspace/genie) to let users ask questions about your data for instant insights (answers and table-like output). You are also able to collect their feedback on the responses. Visualizations aren't yet supported in the API.

## Code snippet

Refer to the Streamlit Cookbook Genie source code for the full implementation.

```python title="app.py"

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.dashboards import GenieFeedbackRating

w = WorkspaceClient()

genie_space_id = "01efe16a65e21836acefb797ae6a8fe4"

def display_message(message):
    if "content" in message:
        st.markdown(message["content"])
    if "data" in message:
        st.dataframe(message["data"])
    if "code" in message:
        with st.expander("Show generated code"):
            st.code(message["code"], language="sql", wrap_lines=True)

def get_query_result(statement_id):
    # For simplicity, let's say data fits in one chunk, query.manifest.total_chunk_count = 1

    result = w.statement_execution.get_statement(statement_id)
    return pd.DataFrame(
        result.result.data_array, columns=[i.name for i in result.manifest.schema.columns]
    )

def collect_feedback(message_id: str):
    rating = st.feedback("thumbs", key=f"feedback_{message_id}")
    mapping = {1: GenieFeedbackRating.POSITIVE, 0: GenieFeedbackRating.NEGATIVE}
    if rating and message["message_id"]:
        w.genie.send_message_feedback(
            genie_space_id, st.session_state.conversation_id, message["message_id"], mapping[rating]
        )

def process_genie_response(response):
    for i in response.attachments:
        if i.text:
            message = {"role": "assistant", "content": i.text.content}
            display_message(message)
        elif i.query:
            data = get_query_result(response.query_result.statement_id)
            message = {
                "role": "assistant", "content": i.query.description, "data": data, "code": i.query.query
            }
            display_message(message)
    collect_feedback(response.message_id)

if prompt := st.chat_input("Ask your question..."):
    # Refer to actual app code for chat history persistence on rerun

    st.chat_message("user").markdown(prompt)

    with st.chat_message("assistant"):
        if st.session_state.get("conversation_id"):
            conversation = w.genie.create_message_and_wait(
                genie_space_id, st.session_state.conversation_id, prompt
            )
            process_genie_response(conversation)
        else:
            conversation = w.genie.start_conversation_and_wait(genie_space_id, prompt)
            process_genie_response(conversation)
```

:::info

Copy and paste the Genie space ID from the Genie UI URL as rooms/SPACE-ID?o=.

:::

## Resources

- [Genie](https://www.databricks.com/what-aibi-genie)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `SELECT` on the Unity Catalog table
- `CAN USE` the SQL warehouse
- `CAN VIEW` the Genie Space

## Dependencies

- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`
- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Pandas](https://pypi.org/project/pandas/) - `pandas`

```python title="requirements.txt"
streamlit
databricks-sdk
pandas
```

---

## Connect to a cluster(3)


This recipe uses [Databricks Connect](https://docs.databricks.com/en/dev-tools/databricks-connect/python/index.html) to execute pre-defined Python or SQL code on a **shared** cluster with UI inputs.

## Code snippet

```python title="app.py"

from databricks.connect import DatabricksSession

cluster_id = st.text_input(
    "Specify cluster id:",
    placeholder="0709-132523-cnhxf2p6",
    help="Copy a shared Compute [cluster ID](https://docs.databricks.com/en/workspace/workspace-details.html#cluster-url-and-id) to connect to.",
)

if cluster_id:
    spark = connect_to_cluster(cluster_id)
    st.success("Successfully connected to Spark:", icon="✅")
    session_info = {
            "App Name": spark.conf.get("spark.app.name", "Unknown"),
            "Master URL": spark.conf.get("spark.master", "Unknown"),
        }
    st.json(session_info)

spark = DatabricksSession.builder.remote(
    host=os.getenv("DATABRICKS_HOST"),
    cluster_id=cluster_id
).getOrCreate()

query = "SELECT 'I'm a stellar cook!' AS message"
sql_output = spark.sql(query).toPandas()

st.dataframe(sql_output)

result = spark.range(10).toPandas()

st.dataframe(result)
```

:::info

You also have the option to [connect to serverless compute using Databricks Connect](https://docs.databricks.com/aws/en/compute/serverless/).

:::

## Resources

- [All-purpose compute](https://docs.databricks.com/aws/en/compute/use-compute) or [serverless compute](https://docs.databricks.com/aws/en/compute/serverless/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN ATTACH TO` permission on the cluster

See [Compute permissions](https://docs.databricks.com/aws/en/compute/clusters-manage#compute-permissions) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`

```python title="requirements.txt"
databricks-sdk
streamlit
```

---

## External connections(3)


This recipe demonstrates how to use Unity Catalog-managed external [HTTP connections](https://docs.databricks.com/aws/en/query-federation/http) for secure and governed access to MCP and non-MCP servers, for example, to GitHub, or Jira, and Slack.

## Code snippets

### OAuth User to Machine Per User (On-behalf-of-user)

```python title="app.py"

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ExternalFunctionRequestHttpMethod

token = st.context.headers.get("x-forwarded-access-token")
w = WorkspaceClient(token=token, auth_type="pat")

response = w.serving_endpoints.http_request(
    conn="github_u2m",
    method=ExternalFunctionRequestHttpMethod.GET,
    path="/user",
    headers={"Accept": "application/vnd.github+json"},
)

st.json(response.json())
```

### Bearer token

```python title="app.py"

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ExternalFunctionRequestHttpMethod

w = WorkspaceClient()

response = w.serving_endpoints.http_request(
    conn="github_connection",
    method=ExternalFunctionRequestHttpMethod.GET,
    path="/traffic/views",
    headers={"Accept": "application/vnd.github+json"},
)

st.json(response.json())
```

## Resources

- [Unity Catalog HTTP Connection](https://docs.databricks.com/aws/en/query-federation/http), either MCP or non-MCP

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CONNECTION` permission on the HTTP Connection

When using OAuth User to Machine Per User (On-behalf-of-user), you need to configure [User authorization](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/auth#user-authorization) by adding the Unity Catalog connection or other scopes.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`

```python title="requirements.txt"
databricks-sdk
streamlit
```

---

## Read a Lakebase table


This app connects to a [Databricks Lakebase](https://docs.databricks.com/aws/en/oltp/) OLTP database instance and reads the first 100 rows from any table. Provide the instance name, database, schema, and table name.

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

def get_connection(host: str, database: str, user: str) -> psycopg.Connection:
    """Get a connection to the Lakebase database using OAuth token."""
    token = w.config.oauth_token().access_token
    
    return psycopg.connect(
        host=host,
        port=5432,
        dbname=database,
        user=user,
        password=token,
        sslmode="require",
    )

def query_df(host: str, database: str, user: str, sql: str) -> pd.DataFrame:
    """Execute a SQL query and return results as a DataFrame."""
    conn = get_connection(host, database, user)
    try:
        with conn.cursor() as cur:
            cur.execute(sql)
            if not cur.description:
                return pd.DataFrame()
            
            cols = [d.name for d in cur.description]
            rows = cur.fetchall()
            return pd.DataFrame(rows, columns=cols)
    finally:
        conn.close()

# Get connection parameters from environment variables (set by Databricks Apps)
# or fall back to manual configuration
host = os.getenv("PGHOST")
database = os.getenv("PGDATABASE")
user = os.getenv("PGUSER")

if not all([host, database, user]):
    # Manual configuration if environment variables are not set
    instance_name = "your_instance_name"
    database = "databricks_postgres"
    user = w.config.client_id or w.current_user.me().user_name
    host = w.database.get_database_instance(name=instance_name).read_write_dns

# Query table
schema = "public"
table = "your_table_name"
df = query_df(host, database, user, f"SELECT * FROM {schema}.{table} LIMIT 100")
st.dataframe(df)
```

:::tip

Add your Lakebase instance as an App resource to automatically configure connection parameters via environment variables. See the [Lakebase resource documentation](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/lakebase) for details.

:::

:::warning

Tokens expire periodically; this app refreshes on each new connection and enforces TLS (sslmode=require).

:::

## Resources

- [Lakebase](https://docs.databricks.com/aws/en/oltp/) database instance (Postgres).
- An existing Postgres database, schema, and table with data.

## Permissions

Add the Lakebase instance as an [**App resource**](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/lakebase) to automatically configure permissions and environment variables (`PGHOST`, `PGDATABASE`, `PGUSER`, etc.).

Alternatively, manually create a Postgres role for the service principal. See [this guide](https://docs.databricks.com/aws/en/oltp/pg-roles?language=PostgreSQL#create-postgres-roles-and-grant-privileges-for-databricks-identities).

Example grants for read access:

```sql
GRANT CONNECT ON DATABASE databricks_postgres TO "<service-principal-id>";
GRANT USAGE ON SCHEMA public TO "<service-principal-id>";
GRANT SELECT ON TABLE your_table_name TO "<service-principal-id>";
```

[This guide](https://docs.databricks.com/aws/en/oltp/query/sql-editor#create-a-new-query) shows you how to query your Lakebase.

## Dependencies

- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk>=0.60.0`
- [Psycopg](https://pypi.org/project/psycopg/) - `psycopg[binary]`
- [Pandas](https://pypi.org/project/pandas/) - `pandas`
- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`

```python title="requirements.txt"
databricks-sdk>=0.60.0
pandas
streamlit
psycopg[binary]
```

---

## Edit a Delta table(3)


Use this recipe to read, edit, and write back data in a [Unity Catalog table](https://docs.databricks.com/aws/en/tables/) using the [Databricks SQL Connector](https://docs.databricks.com/en/dev-tools/python-sql-connector.html).

## Code snippet

```python title="app.py"

from databricks import sql
from databricks.sdk.core import Config

cfg = Config() # Set the DATABRICKS_HOST environment variable when running locally

@st.cache_resource(ttl=300, show_spinner=True)
def get_connection(http_path):
    return sql.connect(
        server_hostname=cfg.host,
        http_path=http_path,
        credentials_provider=lambda: cfg.authenticate,
    )

def read_table(table_name: str, conn) -> pd.DataFrame:
    with conn.cursor() as cursor:
        cursor.execute(f"SELECT * FROM {table_name}")
        return cursor.fetchall_arrow().to_pandas()

def insert_overwrite_table(table_name: str, df: pd.DataFrame, conn):
    progress = st.empty()
    with conn.cursor() as cursor:
        rows = list(df.itertuples(index=False, name=None))
        if not rows:
            return

        cols = list(df.columns)
        num_cols = len(cols)
        params = {}
        values_sql_parts = []
        p = 0
        for row in rows:
            ph = []
            for v in row:
                key = f"p{p}"
                ph.append(f":{key}")
                params[key] = v
                p += 1
            values_sql_parts.append("(" + ",".join(ph) + ")")

        values_sql = ",".join(values_sql_parts)
        col_list_sql = ",".join(cols)
                
        with progress:
            st.info("Calling Databricks SQL...")
        cursor.execute(f"INSERT OVERWRITE {table_name} ({col_list_sql}) VALUES {values_sql}", params)

http_path_input = st.text_input(
    "Specify the HTTP Path to your Databricks SQL Warehouse:",
    placeholder="/sql/1.0/warehouses/xxxxxx",
)

table_name = st.text_input(
    "Specify a Catalog table name:", placeholder="catalog.schema.table"
)

if http_path_input and table_name:
    conn = get_connection(http_path_input)
    original_df = read_table(table_name, conn)
    edited_df = st.data_editor(original_df, num_rows="dynamic", hide_index=True)

    df_diff = pd.concat([original_df, edited_df]).drop_duplicates(keep=False)
    if not df_diff.empty:
        if st.button("Save changes"):
            insert_overwrite_table(table_name, edited_df, conn)
else:
    st.warning("Provide both the warehouse path and a table name to load data.")
```

:::info

This sample uses Streamlit's [st.cache_resource](https://docs.streamlit.io/develop/concepts/architecture/caching#stcache_resource) with a 300-second TTL (time-to-live) to cache the database connection across users, sessions, and reruns. The cached connection will automatically expire after 1 hour, ensuring connections don't become stale. Use Streamlit's caching decorators and TTL parameter to implement a caching strategy that works for your use case.

:::

## Resources

- [SQL warehouse](https://docs.databricks.com/aws/en/compute/sql-warehouse/)
- [Unity Catalog table](https://docs.databricks.com/aws/en/tables/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `MODIFY` on the Unity Catalog table
- `CAN USE` on the SQL warehouse

See Unity [Catalog privileges and securable objects](https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges) for more information.

## Dependencies

- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Databricks SQL Connector](https://pypi.org/project/databricks-sql-connector/) - `databricks-sql-connector`
- [Pandas](https://pypi.org/project/pandas/) - `pandas`
- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`

```python title="requirements.txt"
databricks-sdk
databricks-sql-connector
pandas
streamlit
```

---

## Read a Delta table(3)


This recipe reads a [Unity Catalog table](https://docs.databricks.com/aws/en/tables/) using the [Databricks SQL Connector](https://docs.databricks.com/en/dev-tools/python-sql-connector.html).

## Code snippet

```python title="app.py"

from databricks import sql
from databricks.sdk.core import Config

cfg = Config()  # Set the DATABRICKS_HOST environment variable when running locally

@st.cache_resource(ttl=300, show_spinner=True) # connection is cached
def get_connection(http_path):
    return sql.connect(
        server_hostname=cfg.host,
        http_path=http_path,
        credentials_provider=lambda: cfg.authenticate,
    )

def read_table(table_name, conn):
    with conn.cursor() as cursor:
        query = f"SELECT * FROM {table_name}"
        cursor.execute(query)
        return cursor.fetchall_arrow().to_pandas()

http_path_input = st.text_input(
    "Enter your Databricks HTTP Path:", placeholder="/sql/1.0/warehouses/xxxxxx"
)

table_name = st.text_input(
    "Specify a Unity Catalog table name:", placeholder="catalog.schema.table"
)

if http_path_input and table_name:
    conn = get_connection(http_path_input)
    df = read_table(table_name, conn)
    st.dataframe(df)
```

:::info

This sample uses Streamlit's [st.cache_resource](https://docs.streamlit.io/develop/concepts/architecture/caching#stcache_resource) with a 300-second TTL (time-to-live) to cache the database connection across users, sessions, and reruns. The cached connection will automatically expire after 1 hour, ensuring connections don't become stale. Use Streamlit's caching decorators and TTL parameter to implement a caching strategy that works for your use case.

:::

## Resources

- [SQL warehouse](https://docs.databricks.com/aws/en/compute/sql-warehouse/)
- [Unity Catalog table](https://docs.databricks.com/aws/en/tables/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `SELECT` on the Unity Catalog table
- `CAN USE` on the SQL warehouse

See Unity [Catalog privileges and securable objects](https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges) for more information.

## Dependencies

- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Databricks SQL Connector](https://pypi.org/project/databricks-sql-connector/) - `databricks-sql-connector`
- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`

```python title="requirements.txt"
databricks-sdk
databricks-sql-connector
streamlit
```

---

## Charts


Use this recipe to visualize data using Streamlit's built-in chart components: area charts, line charts, and bar charts. This example demonstrates loading data from a Unity Catalog table and creating various business insights through different chart visualizations.

## Code snippet

### Load data from a table

```python title="app.py"

from databricks import sql
from databricks.sdk.core import Config
from databricks.sdk import WorkspaceClient

cfg = Config()
w = WorkspaceClient()

# List available SQL warehouses
warehouses = w.warehouses.list()
warehouse_paths = {wh.name: wh.odbc_params.path for wh in warehouses}

# Connect to SQL warehouse
@st.cache_resource(ttl=300, show_spinner=True)
def get_connection(http_path):
    return sql.connect(
        server_hostname=cfg.host,
        http_path=http_path,
        credentials_provider=lambda: cfg.authenticate,
    )

# Read table
def read_table(table_name, conn):
    with conn.cursor() as cursor:
        cursor.execute(f"SELECT * FROM {table_name} LIMIT 1000")
        return cursor.fetchall_arrow().to_pandas()

# Get data
warehouse_name = "your_warehouse_name"
table_name = "samples.nyctaxi.trips"

http_path = warehouse_paths[warehouse_name]
conn = get_connection(http_path)
df = read_table(table_name, conn)

# Process datetime columns
df["tpep_pickup_datetime"] = pd.to_datetime(df["tpep_pickup_datetime"])
df["tpep_dropoff_datetime"] = pd.to_datetime(df["tpep_dropoff_datetime"])
df["pickup_hour"] = df["tpep_pickup_datetime"].dt.hour
df["trip_duration_minutes"] = (df["tpep_dropoff_datetime"] - df["tpep_pickup_datetime"]).dt.total_seconds() / 60
```

### Demand analysis: Trips by hour

```python title="app.py"

# Count trips by hour to understand demand patterns
hourly_demand = df["pickup_hour"].value_counts().sort_index()
st.bar_chart(hourly_demand)

peak_hour = hourly_demand.idxmax()
st.info(f"Peak demand hour: {peak_hour}:00 with {hourly_demand.max()} trips")
```

### Revenue analysis: Average fare by hour

```python title="app.py"

# Analyze when fares are highest
avg_fare_by_hour = df.groupby("pickup_hour")["fare_amount"].mean()
st.line_chart(avg_fare_by_hour)

best_hour = avg_fare_by_hour.idxmax()
st.success(f"Best earning hour: {best_hour}:00")
```

### Location analysis: Top pickup zones

```python title="app.py"

# Identify high-demand pickup locations
top_pickups = df["pickup_zip"].value_counts().head(15)
st.bar_chart(top_pickups)
```

### Cumulative revenue over time

```python title="app.py"

# Track total revenue accumulation
revenue_df = df.set_index("tpep_pickup_datetime")[["fare_amount"]].sort_index()
revenue_df["cumulative_revenue"] = revenue_df["fare_amount"].cumsum()
st.area_chart(revenue_df["cumulative_revenue"])
```

## Resources

- [SQL warehouse](https://docs.databricks.com/aws/en/compute/sql-warehouse/)
- [Unity Catalog table](https://docs.databricks.com/aws/en/tables/)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN USE` on the SQL warehouse
- `SELECT` on the Unity Catalog table

See Unity [Catalog privileges and securable objects](https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges) for more information.

## Dependencies

- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`
- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Databricks SQL Connector](https://pypi.org/project/databricks-sql-connector/) - `databricks-sql-connector`
- [Pandas](https://pypi.org/project/pandas/) - `pandas`

```python title="requirements.txt"
streamlit
databricks-sdk
databricks-sql-connector
pandas
```

---

## Map display and interaction


This recipe enables you to display geographic data on a map and collect user geo input through interactive map drawing. You can load location data from Unity Catalog tables or use the drawing tools to capture points, polygons, and geofences from users.

## Code snippet

### Display geo data from a table

```python title="app.py"

from databricks import sql
from databricks.sdk.core import Config
from databricks.sdk import WorkspaceClient

cfg = Config()
w = WorkspaceClient()

# List available SQL warehouses
warehouses = w.warehouses.list()
warehouse_paths = {wh.name: wh.odbc_params.path for wh in warehouses}

# Connect to SQL warehouse
@st.cache_resource(ttl=300, show_spinner=True)
def get_connection(http_path):
    return sql.connect(
        server_hostname=cfg.host,
        http_path=http_path,
        credentials_provider=lambda: cfg.authenticate,
    )

# Read table
def read_table(table_name, conn):
    with conn.cursor() as cursor:
        cursor.execute(f"SELECT * FROM {table_name}")
        return cursor.fetchall_arrow().to_pandas()

# Get data and display on map
warehouse_name = "your_warehouse_name"
table_name = "samples.accuweather.forecast_daily_calendar_metric"

http_path = warehouse_paths[warehouse_name]
conn = get_connection(http_path)
df = read_table(table_name, conn)

# Display map with latitude/longitude columns
st.map(df, latitude="latitude", longitude="longitude")
```

### Collect user geo input

```python title="app.py"

from streamlit_folium import st_folium

from folium.plugins import Draw

# Create a map centered on a location
m = folium.Map(location=[37.7749, -122.4194], zoom_start=13)

# Enable drawing tools (set True for the tools you want to enable)
draw = Draw(
    draw_options={
        "marker": True,      # For collecting points
        "polygon": True,     # For collecting geofences/polygons
        "polyline": True,    # For collecting polylines
        "rectangle": True,   # For collecting rectangles
        "circle": True,      # For collecting circles
        "circlemarker": False,
    },
    edit_options={"edit": True},
)
draw.add_to(m)
output = st_folium(m, width=700, height=500)

# Access the drawn geometry
if output["last_active_drawing"] and "geometry" in output["last_active_drawing"]:
    geometry = output["last_active_drawing"]["geometry"]
    st.json(geometry)
```

## Resources

- [SQL warehouse](https://docs.databricks.com/aws/en/compute/sql-warehouse/) _(optional, only for reading table data)_
- [Unity Catalog table](https://docs.databricks.com/aws/en/tables/) _(optional, only for reading table data)_

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN USE` on the SQL warehouse _(only required if reading data from tables)_
- `SELECT` on the Unity Catalog table _(only required if reading data from tables)_

See Unity [Catalog privileges and securable objects](https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges) for more information.

## Dependencies

- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`
- [Streamlit Folium](https://pypi.org/project/streamlit-folium/) - `streamlit-folium`
- [Databricks SDK](https://pypi.org/project/databricks-sdk/) - `databricks-sdk` _(for table data)_
- [Databricks SQL Connector](https://pypi.org/project/databricks-sql-connector/) - `databricks-sql-connector` _(for table data)_

```python title="requirements.txt"
streamlit
streamlit-folium
databricks-sdk
databricks-sql-connector
```

---

## Download a file(3)


This recipe downloads a file from a [Unity Catalog volume](https://docs.databricks.com/en/volumes/index.html) using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/).

:::note

Unlike notebooks, Databricks Apps does not support mounting Unity Catalog volumes and directly reading and writing files. As this code snippet demonstrates, each file needs to be downloaded to the app compute before being able to manipulate it.

:::

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

download_file_path = st.text_input(
    label="Path to file", placeholder="/Volumes/catalog/schema/volume_name/file.csv"
)

response = w.files.download(download_file_path)
file_data = response.contents.read()
file_name = os.path.basename(download_file_path)

st.download_button(label="Download", data=file_data, file_name=file_name)
```

## Resources

- [Unity Catalog volume](https://docs.databricks.com/aws/en/files/volumes)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CATALOG` on the volume's catalog
- `USE SCHEMA` on the volume's schema
- `READ VOLUME` on the volume

See [Privileges required for volume operations](https://docs.databricks.com/en/volumes/privileges.html#privileges-required-for-volume-operations) for more information.

If you declare volume access in a Databricks Asset Bundle, `resources.apps[*].resources[*].uc_securable` may not grant `USE_CATALOG` and `USE_SCHEMA` on the parent catalog and schema (the app still needs them at runtime). As a temporary workaround until bundles can declare those parent grants, add the privileges manually, or see [apps_grants_sync](https://github.com/salihbout/apps_grants_sync): an example Databricks App and Asset Bundle that wires `experimental.scripts.postdeploy` so parent privileges are applied after each `databricks bundle deploy` (copy its `tools/` into your bundle or mirror the same pattern in `databricks.yml`).

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`

```python title="requirements.txt"
databricks-sdk
streamlit
```

---

## Upload a file(3)


This recipe uploads a file to a [Unity Catalog volume](https://docs.databricks.com/en/volumes/index.html) using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/).

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

uploaded_file = st.file_uploader(label="Select file")

upload_volume_path = st.text_input(
    label="Specify a three-level Unity Catalog volume name (catalog.schema.volume_name)",
    placeholder="main.marketing.raw_files",
)

if st.button("Save changes"):
    file_bytes = uploaded_file.read()
    binary_data = io.BytesIO(file_bytes)
    file_name = uploaded_file.name

    parts = upload_volume_path.strip().split(".")

    catalog = parts[0]
    schema = parts[1]
    volume_name = parts[2]

    volume_file_path = f"/Volumes/{catalog}/{schema}/{volume_name}/{file_name}"
    w.files.upload(volume_file_path, binary_data, overwrite=True)
```

## Resources

- [Unity Catalog volume](https://docs.databricks.com/aws/en/files/volumes)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `USE CATALOG` on the catalog of the volume
- `USE SCHEMA` on the schema of the volume
- `READ VOLUME` and `WRITE VOLUME` on the volume

See [Privileges required for volume operations](https://docs.databricks.com/en/volumes/privileges.html#privileges-required-for-volume-operations) for more information.

If you declare volume access in a Databricks Asset Bundle, `resources.apps[*].resources[*].uc_securable` may not grant `USE_CATALOG` and `USE_SCHEMA` on the parent catalog and schema (the app still needs them at runtime). As a temporary workaround until bundles can declare those parent grants, add the privileges manually, or see [apps_grants_sync](https://github.com/salihbout/apps_grants_sync): an example Databricks App and Asset Bundle that wires `experimental.scripts.postdeploy` so parent privileges are applied after each `databricks bundle deploy` (copy its `tools/` into your bundle or mirror the same pattern in `databricks.yml`).

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`

```python title="requirements.txt"
databricks-sdk
streamlit
```

---

## Retrieve workflow results(3)


This recipe retreives the results of a [Databricks Workflows](https://docs.databricks.com/en/jobs/index.html) job task run using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/)..

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

task_run_id = st.text_input(
    label="Specify a task run ID",
    placeholder="293894477334278",
)

results = w.jobs.get_run_output(task_run_id)

st.text(results)
```

## Resources

- [Job](https://docs.databricks.com/aws/en/jobs/configure-job)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN VIEW` permission on the job

See [Control access to a job](https://docs.databricks.com/en/jobs/privileges.html#control-access-to-a-job) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`

```python title="requirements.txt"
databricks-sdk
streamlit
```

---

## Run a workflow(3)


This recipe triggers a [Databricks Workflows](https://docs.databricks.com/en/jobs/index.html) job using the [Databricks SDK for Python](https://databricks-sdk-py.readthedocs.io/en/latest/).

## Code snippet

```python title="app.py"

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

job_id = st.text_input(
    label="Specify job id:",
    placeholder="921773893211960",
    help="You can find the job ID under job details after opening a job in the UI.",
)

parameters_input = st.text_area(
    label="Specify job parameters as JSON:",
    placeholder='{"param1": "value1", "param2": "value2"}',
)

parameters = eval(parameters_input.strip())

if st.button(label="Trigger job"):
    try:
        run = w.jobs.run_now(job_id=job_id, job_parameters=parameters)
        st.text(f"Started run with ID {run.run_id}")
    except Exception as e:
        st.warning(e)
```

## Resources

- [Job](https://docs.databricks.com/aws/en/jobs/configure-job)

## Permissions

Your [app service principal](https://docs.databricks.com/aws/en/dev-tools/databricks-apps/#how-does-databricks-apps-manage-authorization) needs the following permissions:

- `CAN MANAGE RUN` permission on the job

See [Control access to a job](https://docs.databricks.com/en/jobs/privileges.html#control-access-to-a-job) for more information.

## Dependencies

- [Databricks SDK for Python](https://pypi.org/project/databricks-sdk/) - `databricks-sdk`
- [Streamlit](https://pypi.org/project/streamlit/) - `streamlit`

```python title="requirements.txt"
databricks-sdk
streamlit
```