BlazeSQL Database Connector API Implementation Guide

Overview

This guide describes how to implement a secure API endpoint that enables BlazeSQL to query your database while keeping all query processing within your infrastructure. The API accepts SQL queries, processes them in your environment, and returns the results directly. This functionality can only be used with a BlazeSQL Enterprise subscription.

API Specification

Request

- Method: POST
- Content-Type: application/json
- Note: BlazeSQL will include an API key in the X-API-Key header with each request

Request Body

{
    "query": "SELECT * FROM users LIMIT 10",
    "database_id": "prod_db_01",
    "user_email": "user@company.com"
}

Response Format

On success:

{
    "result": {
        "id": [1, 2, 3],
        "name": ["John", "Jane", "Bob"],
        "created_at": ["2024-02-09T12:00:00Z", "2024-02-09T12:30:00Z", "2024-02-09T13:00:00Z"]
    }
}

On error:

{
    "error": "Invalid SQL syntax near 'SLECT'"
}

Response Codes

200: Success
400: Query Error (invalid syntax, execution error)
404: Database Not Found
500: Miscellaneous Error (server error, connection issues, etc.)

Important Notes

All values in the result must be JSON-serializable:
Numbers (integers, floats): As is
Strings: As is
NULL values: As is
Any other types: Convert to string

Example Implementation

Here's a simple example of how the endpoint might be implemented. This is just for illustration - you can implement the endpoint using any technology stack as long as it meets the API specification above.

from flask import Flask, request, jsonify
from datetime import datetime
from decimal import Decimal
import base64

app = Flask(__name__)

# Example of handling different value types for JSON serialization
def serialize_value(value):
    if value is None:
        return None
    elif isinstance(value, (int, float, str)):
        return value
    elif isinstance(value, datetime):
        return value.isoformat()
    elif isinstance(value, Decimal):
        return str(value)
    elif isinstance(value, bytes):
        return base64.b64encode(value).decode('utf-8')
    else:
        return str(value)

@app.route('/query', methods=['POST'])
def handle_query():
    try:
        # Validate API key
        api_key = request.headers.get('X-API-Key')
        if not api_key or not is_valid_key(api_key):
            return jsonify({'error': 'Invalid API key'}), 401

        # Get request data
        data = request.get_json()
        query = data.get('query')
        database_id = data.get('database_id')
        user_email = data.get('user_email')

        # Check if database exists
        if not database_exists(database_id):
            return jsonify({'error': 'Database not found'}), 404

        # Execute query and get results
        results = execute_query(query, database_id)
        
        # Format results by column
        formatted_results = {}
        for column in results.columns:
            formatted_results[column] = [
                serialize_value(value) for value in results[column]
            ]

        return jsonify({'result': formatted_results}), 200

    except QueryError as e:
        return jsonify({'error': str(e)}), 400
    except Exception as e:
        return jsonify({'error': 'Internal server error'}), 500

if __name__ == '__main__':
    app.run()

Updated on: 28/02/2025