12. BI Tool Integration

Business Intelligence tools aggregate data for visualization. Integrating local AI with BI platforms requires understanding their data models, query interfaces, and extension points.

Direct Database Integration

Many BI tools connect directly to databases. Configure connections with appropriate read permissions.

-- Create dedicated read-only user for BI integration
CREATE USER bi_reader WITH PASSWORD 'secure_password_here';
GRANT CONNECT ON DATABASE production TO bi_reader;
GRANT USAGE ON SCHEMA analytics TO bi_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA analytics TO bi_reader;

-- For tables with sensitive data, create views with limited columns
CREATE VIEW bi_reader.customer_summary AS
SELECT 
    customer_id,
    signup_date,
    tier,
    lifetime_value
FROM customers
WHERE deleted_at IS NULL;

REST API Integration

For BI tools supporting custom data sources, implement API endpoints.

# bi_data_server.py
from flask import Flask, jsonify, request
import ollama
import json
from datetime import datetime
from pathlib import Path

app = Flask(__name__)

# Cache for expensive queries
query_cache = {}
CACHE_TTL_SECONDS = 300

@app.route("/api/v1/metrics/<metric_name>")
def get_metric(metric_name):
    """Return predefined metrics for BI tools."""
    cache_key = f"{metric_name}_{request.args.get('period', 'day')}"
    
    # Check cache
    if cache_key in query_cache:
        cached, timestamp = query_cache[cache_key]
        if datetime.now().timestamp() - timestamp < CACHE_TTL_SECONDS:
            return jsonify(cached)
    
    # Compute metric
    result = compute_metric(metric_name, request.args)
    
    # Cache result
    query_cache[cache_key] = (result, datetime.now().timestamp())
    
    return jsonify(result)

def compute_metric(metric_name: str, params: dict) -> dict:
    """Compute metrics using local AI for complex aggregations."""
    period = params.get("period", "day")
    
    if metric_name == "anomaly_score":
        return compute_anomaly_score(period)
    elif metric_name == "trend_forecast":
        return compute_forecast(period)
    elif metric_name == "segment_insight":
        return generate_segment_insight(params)
    else:
        return {"error": f"Unknown metric: {metric_name}"}

def compute_anomaly_score(period: str) -> dict:
    """Use local LLM to detect anomalies in data."""
    # Load recent data
    data = load_recent_data(period)
    
    prompt = f"""Analyze this data for anomalies:
{data}

Return JSON with:
- "anomalies": list of anomaly records with timestamp, metric, value, expected_range
- "summary": one-sentence description of overall health
- "risk_score": 0-100 rating
"""
    
    response = ollama.chat(
        model="llama3",
        messages=[{"role": "user", "content": prompt}],
        options={"temperature": 0.1}
    )
    
    # Parse and return structured result
    try:
        return json.loads(response["message"]["content"])
    except json.JSONDecodeError:
        return {"error": "Failed to parse AI response", "raw": response}

@app.route("/api/v1/query", methods=["POST"])
def custom_query():
    """Execute custom analytical queries."""
    query_request = request.json
    
    if not query_request.get("query"):
        return jsonify({"error": "Query required"}), 400
    
    # Validate query safety
    if any(unsafe in query_request["query"].lower() for unsafe in 
           ["drop ", "delete ", "truncate ", "alter ", "insert ", "update "]):
        return jsonify({"error": "Only SELECT queries allowed"}), 403
    
    result = execute_read_query(query_request["query"])
    return jsonify(result)

def execute_read_query(query: str) -> dict:
    """Execute a read-only query against the data warehouse."""
    # Implementation depends on your database
    pass

@app.route("/health")
def health():
    """Health check endpoint for monitoring."""
    return jsonify({
        "status": "healthy",
        "cache_size": len(query_cache),
        "model_available": check_ollama_connection()
    })

def check_ollama_connection() -> bool:
    """Verify Ollama service is available."""
    try:
        ollama.list()
        return True
    except Exception:
        return False

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

Tableau and Power BI Integration

For Tableau Extract API or Power BI REST API, use SDKs:

# tableau_integration.py
from tableauscraper import TableauScraper
import json

class TableauConnector:
    def __init__(self, server_url: str, site: str = ""):
        self.server_url = server_url
        self.site = site
        self.ts = None
    
    def authenticate(self, api_token: str):
        """Authenticate using personal access token."""
        self.ts = TableauScraper()
        self.ts.authenticate(
            server=self.server_url,
            token_name="automation_user",
            token_value=api_token,
            site=self.site
        )
    
    def get_workbook_data(self, workbook_name: str, view_name: str) -> list[dict]:
        """Extract data from a specific view."""
        url = f"{self.server_url}/views/{workbook_name}/{view_name}"
        self.ts.loads(url)
        
        worksheet = self.ts.getWorksheet(view_name)
        return worksheet.data

# Power BI integration using REST API
# power_bi_integration.py
import requests

class PowerBIClient:
    def __init__(self, tenant_id: str, client_id: str, client_secret: str):
        self.tenant_id = tenant_id
        self.client_id = client_id
        self.client_secret = client_secret
        self.access_token = self._get_access_token()
    
    def _get_access_token(self) -> str:
        """Obtain OAuth access token."""
        url = f"https://login.microsoftonline.com/{self.tenant_id}/oauth2/v2.0/token"
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "https://analysis.windows.net/powerbi/api/.default"
        }
        response = requests.post(url, data=data)
        return response.json()["access_token"]
    
    def get_datasets(self, workspace_id: str) -> list[dict]:
        """List datasets in a workspace."""
        url = f"https://api.powerbi.com/v1.0/myorg/groups/{workspace_id}/datasets"
        headers = {"Authorization": f"Bearer {self.access_token}"}
        response = requests.get(url, headers=headers)
        return response.json()["value"]
    
    def refresh_dataset(self, workspace_id: str, dataset_id: str):
        """Trigger dataset refresh."""
        url = f"https://api.powerbi.com/v1.0/myorg/groups/{workspace_id}/datasets/{dataset_id}/refreshes"
        headers = {"Authorization": f"Bearer {self.access_token}"}
        requests.post(url, headers=headers)