Storage Access

import json
import os
from keboola_query_service import Client

# Storage Access config is set by the platform when the feature is enabled.
# workspace_id is read from the manifest file (recommended); the other values
# are plain env vars.
try:
    branch_id = os.environ["BRANCH_ID"]
    query_service_url = os.environ["QUERY_SERVICE_URL"]
    with open(os.environ["KBC_WORKSPACE_MANIFEST_PATH"]) as f:
        workspace_id = json.load(f)["workspaceId"]
except (KeyError, FileNotFoundError) as e:
    raise RuntimeError(
        "Storage Access is not enabled for this app. "
        "Enable it in Advanced Settings and redeploy."
    ) from e

# Initialize the Query Service client
client = Client(
    base_url=query_service_url,
    token=os.environ["KBC_TOKEN"],
)

import pandas as pd

# Query a table - use the full table ID (bucket.table)
results = client.execute_query(
    branch_id=branch_id,
    workspace_id=workspace_id,
    statements=['SELECT * FROM "in.c-main"."customers" LIMIT 1000'],
)

# One QueryResult per statement — we sent one statement, so take the first.
result = results[0]

# Convert to DataFrame
df = pd.DataFrame(result.data, columns=[c.name for c in result.columns])
print(df.head())

# Join multiple tables
query = """
    SELECT
        c.customer_name,
        SUM(o.amount) as total_spent
    FROM "in.c-main"."customers" c
    JOIN "in.c-main"."orders" o ON c.id = o.customer_id
    GROUP BY c.customer_name
    ORDER BY total_spent DESC
    LIMIT 10
"""

results = client.execute_query(
    branch_id=branch_id,
    workspace_id=workspace_id,
    statements=[query],
)
result = results[0]

# INSERT new records
client.execute_query(
    branch_id=branch_id,
    workspace_id=workspace_id,
    statements=['''
        INSERT INTO "in.c-main"."approvals" ("id", "name", "status", "updated_at")
        VALUES (1, 'New Record', 'pending', CURRENT_TIMESTAMP)
    '''],
)

# UPDATE existing records
client.execute_query(
    branch_id=branch_id,
    workspace_id=workspace_id,
    statements=['''
        UPDATE "in.c-main"."approvals"
        SET status = 'approved', updated_at = CURRENT_TIMESTAMP
        WHERE id = 123
    '''],
)

# DELETE records
client.execute_query(
    branch_id=branch_id,
    workspace_id=workspace_id,
    statements=['''
        DELETE FROM "in.c-main"."approvals"
        WHERE status = 'cancelled'
    '''],
)

client.execute_query(
    branch_id=branch_id,
    workspace_id=workspace_id,
    statements=['TRUNCATE TABLE "in.c-main"."temp_data"'],
)

# Example: Optimistic locking with version column
query = """
    UPDATE "in.c-main"."records"
    SET status = 'approved', version = version + 1
    WHERE id = 123 AND version = 5
"""
results = client.execute_query(
    branch_id=branch_id,
    workspace_id=workspace_id,
    statements=[query],
)

if results[0].rows_affected == 0:
    raise Exception("Record was modified by another user. Please refresh and try again.")

import json
import os

try:
    branch_id = os.environ["BRANCH_ID"]
    query_service_url = os.environ["QUERY_SERVICE_URL"]
    with open(os.environ["KBC_WORKSPACE_MANIFEST_PATH"]) as f:
        workspace_id = json.load(f)["workspaceId"]
except (KeyError, FileNotFoundError) as e:
    raise RuntimeError(
        "Storage Access is not enabled. Enable it in Advanced Settings and redeploy."
    ) from e

from flask import Flask, request, render_template_string
import json
import os
from keboola_query_service import Client

app = Flask(__name__)

# Read Storage Access config once at startup.
# workspace_id is read from the manifest (recommended); other values are env vars.
BRANCH_ID = os.environ["BRANCH_ID"]
with open(os.environ["KBC_WORKSPACE_MANIFEST_PATH"]) as f:
    WORKSPACE_ID = json.load(f)["workspaceId"]

qs_client = Client(
    base_url=os.environ["QUERY_SERVICE_URL"],
    token=os.environ["KBC_TOKEN"],
)

ALLOWED_STATUSES = {"pending", "approved", "rejected"}


@app.route("/", methods=["GET", "POST"])
def index():
    if request.method == "POST":
        # Validate and sanitize user input BEFORE it reaches SQL.
        # int() guarantees record_id is a number; the allowlist guarantees
        # new_status is one of three exact strings. This is the only reason
        # the f-string below is safe — do not add other form fields to the
        # query without analogous validation.
        record_id = int(request.form["record_id"])
        new_status = request.form["status"]
        if new_status not in ALLOWED_STATUSES:
            return "Invalid status", 400

        qs_client.execute_query(
            branch_id=BRANCH_ID,
            workspace_id=WORKSPACE_ID,
            statements=[f'''
                UPDATE "in.c-main"."approvals"
                SET status = '{new_status}', updated_at = CURRENT_TIMESTAMP
                WHERE id = {record_id}
            '''],
        )

    # Load current records
    results = qs_client.execute_query(
        branch_id=BRANCH_ID,
        workspace_id=WORKSPACE_ID,
        statements=['SELECT id, name, status, updated_at FROM "in.c-main"."approvals" ORDER BY id'],
    )
    result = results[0]
    column_names = [c.name for c in result.columns]
    records = [dict(zip(column_names, row)) for row in result.data]

    return render_template_string(TEMPLATE, records=records)


TEMPLATE = """
<!DOCTYPE html>
<html>
<head><title>Approval Manager</title></head>
<body>
    <h1>Pending Approvals</h1>
    <table border="1">
        <tr><th>ID</th><th>Name</th><th>Status</th><th>Action</th></tr>
        
    </table>
</body>
</html>
"""

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

import json
import os

try:
    branch_id = os.environ["BRANCH_ID"]
    query_service_url = os.environ["QUERY_SERVICE_URL"]
    with open(os.environ["KBC_WORKSPACE_MANIFEST_PATH"]) as f:
        workspace_id = json.load(f)["workspaceId"]
except (KeyError, FileNotFoundError):
    # Storage Access is not enabled — show a user-friendly error
    import streamlit as st  # or use your framework's error handling
    st.error("Storage Access is not enabled for this app. Enable it in Advanced Settings and redeploy.")
    st.stop()

# ❌ DANGEROUS - never do this
query = f"SELECT * FROM table WHERE id = {user_input}"

# ✅ SAFE - validate types and use allowlists
safe_id = int(user_input)  # Ensure it's actually a number
query = f"SELECT * FROM table WHERE id = {safe_id}"

# ✅ For string values, use an allowlist of permitted values
ALLOWED_STATUSES = {"pending", "approved", "rejected"}
if status not in ALLOWED_STATUSES:
    raise ValueError(f"Invalid status: {status}")
query = f"UPDATE table SET status = '{status}' WHERE id = {safe_id}"

page_size = 1000
last_id = 0  # Start from the beginning

while True:
    results = client.execute_query(
        branch_id=branch_id,
        workspace_id=workspace_id,
        statements=[f'''
            SELECT * FROM "in.c-main"."my_table"
            WHERE id > {last_id}
            ORDER BY id ASC
            LIMIT {page_size}
        '''],
    )
    rows = results[0].data
    if not rows:
        break
    process_batch(rows)
    last_id = rows[-1][0]  # Update cursor to last row's id

import streamlit as st

@st.cache_data(ttl=300)  # Cache for 5 minutes
def load_reference_data():
    results = client.execute_query(
        branch_id=branch_id,
        workspace_id=workspace_id,
        statements=['SELECT * FROM "in.c-main"."reference_data"'],
    )
    result = results[0]
    return pd.DataFrame(result.data, columns=[c.name for c in result.columns])

import time

_cache = {}
_cache_ttl = 300  # seconds

def get_cached_data(key, query_fn):
    now = time.time()
    if key in _cache and now - _cache[key]["ts"] < _cache_ttl:
        return _cache[key]["data"]
    data = query_fn()
    _cache[key] = {"data": data, "ts": now}
    return data

import logging
logging.basicConfig(level=logging.INFO)

logging.info(f"User {current_user} updated record {record_id} to status {new_status}")
# Output goes to stdout → visible in the Terminal Log tab of your Data App