1. Using `select_ai` with Python web frameworks¶

Python web applications should create a Select AI connection pool when the application starts and close it when the application shuts down. A pool lets concurrent requests share a bounded set of database connections instead of creating standalone connections per request.

This pattern works with Python Web Server Gateway Interface (WSGI) and Asynchronous Server Gateway Interface (ASGI) frameworks. FastAPI is used below as a concrete example, but the same approach applies to frameworks such as Flask, Django, Starlette, Sanic, and Quart: initialize the pool during application startup, use select_ai APIs inside request handlers, and close the pool during application shutdown.

Do not call select_ai.connect() or select_ai.create_pool() inside every request handler. Creating connections per request adds latency, increases database connection churn, and can exhaust the database connection limit under load. Create one pool per worker process and let select_ai acquire and release connections from that pool for each API call.

For background and concurrency measurements, see this connection pooling blog.

1.1. Framework patterns¶

Use the framework lifecycle API that runs once per process:

FastAPI / Starlette: use a lifespan async context manager. Create the pool before yield and close it after yield.
Flask: create the pool in create_app(). Close the pool from the shutdown hook provided by the process that runs Flask, such as a Gunicorn worker_exit hook. For simple local applications, atexit.register() can be used for normal interpreter shutdown.
Django: create the pool in AppConfig.ready(). Close the pool from the shutdown hook provided by the process that runs Django, such as a Gunicorn worker_exit hook. For simple local applications, atexit.register() can be used for normal interpreter shutdown.
Quart: use @app.before_serving / @app.after_serving, or @app.while_serving with cleanup after yield.
Sanic: use @app.before_server_start and @app.after_server_stop.

Use select_ai.create_pool() for synchronous request handlers and select_ai.create_pool_async() for asynchronous request handlers. In general, synchronous routes should call synchronous Select AI methods, and async routes should call async Select AI methods. Avoid mixing blocking synchronous database calls into async routes unless the framework runs them in a worker thread.

For Flask and Django, be careful with hooks that run per request or per application context. A Select AI pool should live for the worker process, not for a single request. For example, Flask’s teardown_appcontext runs when an application context is popped, so it is not a good place to close a process-wide pool after every request.

1.2. Install dependencies¶

Install select_ai and FastAPI server dependencies:

python -m pip install select_ai fastapi uvicorn

For local development, set the database connection details as environment variables:

export SELECT_AI_USER=<select_ai_db_user>
export SELECT_AI_PASSWORD=<select_ai_db_password>
export SELECT_AI_DB_CONNECT_STRING=<db_connect_string>
export SELECT_AI_POOL_MIN=5
export SELECT_AI_POOL_MAX=10
export SELECT_AI_POOL_INCREMENT=5

If you use an mTLS wallet, also set TNS_ADMIN or pass wallet parameters to select_ai.create_pool() / select_ai.create_pool_async().

For production deployments, store these values in your deployment platform’s secret manager or environment configuration. Do not hard-code database passwords, wallet passwords, or provider credentials in application source.

1.3. FastAPI synchronous endpoints¶

Create a file named app.py:

import os
from contextlib import asynccontextmanager

from fastapi import FastAPI

import select_ai

user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")

pool_min = int(os.getenv("SELECT_AI_POOL_MIN", "5"))
pool_max = int(os.getenv("SELECT_AI_POOL_MAX", "10"))
pool_increment = int(os.getenv("SELECT_AI_POOL_INCREMENT", "5"))


@asynccontextmanager
async def lifespan(app: FastAPI):
    select_ai.create_pool(
        user=user,
        password=password,
        dsn=dsn,
        min_size=pool_min,
        max_size=pool_max,
        increment=pool_increment,
    )
    yield
    select_ai.disconnect()


app = FastAPI(lifespan=lifespan)


@app.get("/chat")
def chat(prompt: str):
    profile = select_ai.Profile(profile_name="oci_ai_profile")
    return {"response": profile.chat(prompt=prompt)}


@app.get("/show_sql")
def show_sql(prompt: str):
    profile = select_ai.Profile(profile_name="oci_ai_profile")
    return {"sql": profile.show_sql(prompt=prompt)}

Start the server:

uvicorn app:app --host 0.0.0.0 --port 8000

Call the service:

curl "http://localhost:8000/chat?prompt=What%20is%20OCI%3F"

Stop the server by pressing Ctrl+C in the terminal where uvicorn is running. FastAPI runs the lifespan shutdown hook and select_ai.disconnect() closes the pool.

This example creates the Profile proxy inside each handler. The proxy is lightweight; the database connection is acquired from the pool only when the profile method calls the database.

1.4. FastAPI asynchronous endpoints¶

For async endpoints, initialize the async pool with select_ai.create_pool_async() and close it with select_ai.async_disconnect().

import os
from contextlib import asynccontextmanager

from fastapi import FastAPI

import select_ai

user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")


@asynccontextmanager
async def lifespan(app: FastAPI):
    select_ai.create_pool_async(
        user=user,
        password=password,
        dsn=dsn,
        min_size=5,
        max_size=10,
        increment=5,
    )
    yield
    await select_ai.async_disconnect()


app = FastAPI(lifespan=lifespan)


@app.get("/chat")
async def chat(prompt: str):
    profile = await select_ai.AsyncProfile(
        profile_name="async_oci_ai_profile"
    )
    return {"response": await profile.chat(prompt=prompt)}

Start and stop the server the same way:

uvicorn app:app --host 0.0.0.0 --port 8000

Press Ctrl+C to stop it.

1.5. Flask example¶

Flask applications normally use synchronous request handlers, so initialize a synchronous Select AI pool when the application is created.

import atexit
import os

from flask import Flask, jsonify, request

import select_ai


def create_app():
    app = Flask(__name__)

    select_ai.create_pool(
        user=os.getenv("SELECT_AI_USER"),
        password=os.getenv("SELECT_AI_PASSWORD"),
        dsn=os.getenv("SELECT_AI_DB_CONNECT_STRING"),
        min_size=int(os.getenv("SELECT_AI_POOL_MIN", "1")),
        max_size=int(os.getenv("SELECT_AI_POOL_MAX", "4")),
        increment=int(os.getenv("SELECT_AI_POOL_INCREMENT", "1")),
    )

    @app.get("/show_sql")
    def show_sql():
        prompt = request.args["prompt"]
        profile = select_ai.Profile(profile_name="oci_ai_profile")
        return jsonify({"sql": profile.show_sql(prompt=prompt)})

    @atexit.register
    def close_select_ai_pool():
        if select_ai.is_connected():
            select_ai.disconnect()

    return app

1.6. Django example¶

Django has its own database connection management, but select_ai uses the Oracle Database connection pool created by this package. Create the Select AI pool once per process, then call Profile APIs inside views.

# myapp/apps.py
import os

from django.apps import AppConfig

import select_ai


class MyAppConfig(AppConfig):
    name = "myapp"

    def ready(self):
        if not select_ai.is_connected():
            select_ai.create_pool(
                user=os.getenv("SELECT_AI_USER"),
                password=os.getenv("SELECT_AI_PASSWORD"),
                dsn=os.getenv("SELECT_AI_DB_CONNECT_STRING"),
                min_size=int(os.getenv("SELECT_AI_POOL_MIN", "1")),
                max_size=int(os.getenv("SELECT_AI_POOL_MAX", "4")),
                increment=int(os.getenv("SELECT_AI_POOL_INCREMENT", "1")),
            )

# myapp/views.py
from django.http import JsonResponse

import select_ai


def show_sql(request):
    profile = select_ai.Profile(profile_name="oci_ai_profile")
    sql = profile.show_sql(prompt=request.GET["prompt"])
    return JsonResponse({"sql": sql})

When using Django’s development autoreloader, startup hooks may run more than once. The is_connected() check prevents this example from creating a second pool in the same process.

1.7. Pool sizing¶

Use connection pooling for concurrent services such as API applications, workloads with mixed fast and slow requests, and applications with tail-latency requirements. Use standalone connections for simple scripts, command-line tools, or low-concurrency batch jobs.

Set pool sizing based on expected request concurrency and database capacity. In multi-worker deployments, each worker process creates its own pool, so total possible database connections are approximately:

workers * SELECT_AI_POOL_MAX

Choose pool sizes that leave capacity for other database clients and avoid overwhelming small database deployments.

For example, a service running four worker processes with SELECT_AI_POOL_MAX=10 can open up to forty Select AI database connections. If the database can only spare twenty connections for the application, use fewer workers, reduce SELECT_AI_POOL_MAX, or both.

1.8. Pool wait behavior¶

select_ai.create_pool() and select_ai.create_pool_async() pass pool options through to python-oracledb. Use wait_timeout and getmode to control what happens when all pooled connections are busy.

select_ai.create_pool(
    user=user,
    password=password,
    dsn=dsn,
    min_size=2,
    max_size=8,
    increment=2,
    wait_timeout=10,
)

Choose a timeout that matches your API latency budget. For public HTTP APIs, it is usually better to fail fast and return an application error than to let requests pile up until every worker is blocked.

1.9. Request handling¶

Validate prompts and profile names before calling Select AI methods. If clients can choose a profile, check the requested profile against an application allowlist instead of passing arbitrary user input directly into Profile(profile_name=...).

For long-running prompts, set HTTP server timeouts and client timeouts deliberately. Text generation and RAG calls can take longer than simple SQL metadata operations, especially when external AI providers or object storage are involved.

For streaming responses, prefer async frameworks and async Select AI methods when the rest of the application is already async. For ordinary JSON responses, either synchronous or asynchronous routes are fine as long as the connection pool matches the route style.

1. Using `select_ai` with Python web frameworks¶

1.1. Framework patterns¶

1.2. Install dependencies¶

1.3. FastAPI synchronous endpoints¶

1.4. FastAPI asynchronous endpoints¶

1.5. Flask example¶

1.6. Django example¶

1.7. Pool sizing¶

1.8. Pool wait behavior¶

1.9. Request handling¶

Table of Contents

Previous topic

Next topic

This Page

1. Using select_ai with Python web frameworks¶

1.1. Framework patterns¶

1.2. Install dependencies¶

1.3. FastAPI synchronous endpoints¶

1.4. FastAPI asynchronous endpoints¶

1.5. Flask example¶

1.6. Django example¶

1.7. Pool sizing¶

1.8. Pool wait behavior¶

1.9. Request handling¶

1. Using `select_ai` with Python web frameworks¶