はじめに

この記事では、Oracle Cloud Infrastructure (OCI) の Autonomous Database (ADB) に搭載された Select AI 機能を、Python で REST API を公開する方法について説明します。Flask を使用して REST API を作成し、Oracle Database に接続して Select AI クエリを実行します。

必要なパッケージのインストール

まず、必要な Python パッケージをインストールします。以下のコマンドを実行してください。

pip install Flask oracledb werkzeug python-dotenv

アプリケーションの作成

次に、app.py ファイルを作成します。このファイルは、Flask アプリケーションの設定やデータベース接続、エンドポイントの定義などを含みます。

ファイル構造

project/
├── app.py
├── .env
└── app.log

`.env` ファイルの作成

環境変数を管理するために .env ファイルを作成します。以下の内容を記述してください。

ORACLE_23AI_CONNECTION_STRING=your_oracle_user/your_oracle_password@your_adb_high
ORACLE_CLIENT_LIB_DIR=/u01/aipoc/instantclient_23_5
HOST=0.0.0.0
PORT=9999
DEBUG=False

`app.py` の内容

以下は、app.py の内容です。このファイルは、Flask アプリケーションの設定やデータベース接続、エンドポイントの定義などを含みます。

from flask import Flask, request, jsonify
import oracledb
import os
import platform
from dotenv import load_dotenv
from functools import wraps
import logging
import time
from werkzeug.middleware.proxy_fix import ProxyFix
from typing import List, Dict, Any, Callable, Optional, Union

# Initialize Flask application
app = Flask(__name__)
app.wsgi_app = ProxyFix(app.wsgi_app)

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[logging.StreamHandler(), logging.FileHandler('app.log')]
)
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()

# Initialize Oracle client (required for Linux systems)
if platform.system() == 'Linux':
    try:
        oracledb.init_oracle_client(lib_dir=os.getenv("ORACLE_CLIENT_LIB_DIR"))
        logger.info("Oracle client initialized successfully")
    except Exception as e:
        logger.error(f"Failed to initialize Oracle client: {str(e)}")
        raise

# Create database connection pool
pool = None
try:
    pool = oracledb.create_pool(
        dsn=os.getenv("ORACLE_23AI_CONNECTION_STRING"),
        min=2,
        max=10,
        increment=1,
        timeout=30,
        getmode=oracledb.POOL_GETMODE_WAIT,  # Wait for available connection
        wait_timeout=10000,  # Wait timeout in milliseconds
        max_lifetime_session=3600,  # Max lifetime of a connection in seconds
    )
    logger.info("Successfully created Oracle connection pool")
except Exception as e:
    logger.error(f"Failed to create Oracle connection pool: {str(e)}")
    raise


def validate_request_data(required_fields: List[str]) -> Callable:
    """Decorator to validate request data"""

    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            try:
                data = request.get_json()
                if not data:
                    return jsonify(error="Missing request body"), 400

                missing_fields = [field for field in required_fields if field not in data or not data[field]]
                if missing_fields:
                    return jsonify(error=f"Missing required fields: {', '.join(missing_fields)}"), 400

                # Sanitize inputs
                for field in data:
                    if isinstance(data[field], str):
                        data[field] = data[field].strip()

                kwargs['data'] = data
                return f(*args, **kwargs)
            except Exception as e:
                logger.error(f"Error validating request data: {str(e)}")
                return jsonify(error="Invalid request format"), 400

        return decorated_function

    return decorator


def db_connection(f):
    """Decorator to handle database connection with retry logic"""

    @wraps(f)
    def wrapper(*args, **kwargs):
        conn = None
        retries = 3
        retry_delay = 0.5  # seconds

        for attempt in range(retries):
            try:
                conn = pool.acquire()
                kwargs['conn'] = conn
                return f(*args, **kwargs)
            except oracledb.DatabaseError as e:
                error_obj, = e.args
                # Only retry on connection errors, not on SQL errors
                if error_obj.code in (-3113, -3114, -1034, -1012):  # Connection-related error codes
                    logger.warning(f"Database connection error (attempt {attempt + 1}/{retries}): {str(e)}")
                    if attempt < retries - 1:
                        time.sleep(retry_delay)
                        retry_delay *= 2  # Exponential backoff
                        continue
                logger.error(f"Database error: {str(e)}")
                return jsonify(error="Database operation failed"), 500
            except Exception as e:
                logger.error(f"Unexpected error: {str(e)}")
                return jsonify(error="Internal server error"), 500
            finally:
                if conn:
                    try:
                        pool.release(conn)
                    except Exception as e:
                        logger.error(f"Error releasing connection: {str(e)}")

        return jsonify(error="Failed to connect to database after multiple attempts"), 503

    return wrapper


def process_oracle_result(item: Any) -> Any:
    """Process Oracle result items to make them JSON serializable"""
    if hasattr(item, 'read'):  # Handle LOB objects
        return item.read()
    return item


def log_request_info(action: str, query: str) -> None:
    """Log request information with proper truncation"""
    max_length = 100
    truncated_query = query[:max_length] + '...' if len(query) > max_length else query
    logger.info(f"Processing {action} request: {truncated_query}")


@app.route('/selectai/<action>', methods=['POST'])
@validate_request_data(['query'])
@db_connection
def ai_query(action: str, conn: oracledb.Connection, data: Dict[str, Any]) -> tuple:
    """Handle AI query requests"""
    try:
        query = data['query']
        log_request_info(action, query)

        with conn.cursor() as cursor:
            if action == 'showsql':
                cursor.execute("""
                    SELECT DBMS_CLOUD_AI.GENERATE(
                        prompt => :1,
                        profile_name => 'OCI_GENAI',
                        action => 'showsql'
                    ) FROM dual
                """, [query])

                result = cursor.fetchone()
                if result and result[0]:
                    generated_sql = process_oracle_result(result[0])
                    return jsonify(sql=generated_sql)
                return jsonify(error="No SQL generated"), 404

            elif action == 'runsql':
                sql = query.replace('\\"', '"')

                # Execute the query with proper error handling
                try:
                    cursor.execute(sql)

                    # Process results
                    columns = [col[0] for col in cursor.description]
                    rows = [[process_oracle_result(item) for item in row] for row in cursor.fetchall()]

                    row_count = len(rows)
                    logger.info(f"Query returned {row_count} rows")

                    return jsonify(
                        columns=columns,
                        data=rows,
                        row_count=row_count
                    )
                except oracledb.DatabaseError as e:
                    error_obj, = e.args
                    logger.error(f"SQL execution error: {error_obj.message}")
                    return jsonify(error=f"SQL execution error: {error_obj.message}"), 400

            else:
                return jsonify(error=f"Invalid action: {action}"), 400

    except Exception as e:
        logger.error(f"Error processing request: {str(e)}")
        return jsonify(error=str(e)), 500


@app.route('/health', methods=['GET'])
def health_check() -> tuple:
    """Health check endpoint with detailed status information"""
    start_time = time.time()
    try:
        status = {
            'status': 'ok',
            'database': False,
            'pool_stats': {
                'busy': pool.busy if pool else 0,
                'opened': pool.opened if pool else 0,
                'max': pool.max if pool else 0
            },
            'timestamp': int(start_time)
        }

        if pool:
            try:
                # Test database connection
                conn = pool.acquire()
                with conn.cursor() as cursor:
                    cursor.execute("SELECT 1 FROM dual")
                    cursor.fetchone()
                pool.release(conn)
                status['database'] = True
                status['response_time_ms'] = int((time.time() - start_time) * 1000)
            except Exception as e:
                logger.warning(f"Database health check failed: {str(e)}")
                status['status'] = 'degraded'
                status['error'] = str(e)
        else:
            status['status'] = 'critical'

        return jsonify(status), 200 if status['status'] == 'ok' else 503

    except Exception as e:
        logger.error(f"Health check failed: {str(e)}")
        response_time = int((time.time() - start_time) * 1000)
        return jsonify(
            status='error',
            error=str(e),
            response_time_ms=response_time
        ), 500


if __name__ == '__main__':
    app.run(
        host=os.getenv('HOST', '0.0.0.0'),
        port=int(os.getenv('PORT', 9999)),
        debug=os.getenv('DEBUG', 'False').lower() == 'true'
    )

アプリケーションの実行

アプリケーションを実行するために、以下のコマンドを実行します。

python app.py

テスト

以下の curl コマンドを使用して、API の動作を確認します。

`showsql` エンドポイントのテスト

curl --location --request POST 'http://127.0.0.1:9999/selectai/showsql' \
--header 'Content-Type: application/json' \
--data-raw '{
    "query": "how many regions exist"
}'

`showsql` エンドポイントの結果

{
  "sql": "SELECT COUNT(\"r\".\"REGION_ID\") AS \"REGION_COUNT\" FROM \"ADMIN\".\"REGIONS\" \"r\""
}

`runsql` エンドポイントのテスト

curl --location --request POST 'http://127.0.0.1:9999/selectai/runsql' \
--header 'Content-Type: application/json' \
--data-raw '{
    "query": "SELECT COUNT(\"r\".\"REGION_ID\") AS \"REGION_COUNT\" FROM \"ADMIN\".\"REGIONS\" \"r\""
}'

`runsql` エンドポイントの結果

{
  "columns": [
    "REGION_COUNT"
  ],
  "data": [
    [
      5
    ]
  ],
  "row_count": 1
}

まとめ

この記事では、OCI ADB Select AI を Python で REST API に公開する方法について説明しました。Flask を使用して API を構築し、Oracle Database に接続して Select AI クエリを実行することで、Oracle Cloud Infrastructure の Select AI 機能を効率的に活用することができます。