.. _web_frameworks: ************************************************** 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 `__. 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. Install dependencies ==================== Install ``select_ai`` and FastAPI server dependencies: .. code-block:: sh python -m pip install select_ai fastapi uvicorn For local development, set the database connection details as environment variables: .. code-block:: sh export SELECT_AI_USER= export SELECT_AI_PASSWORD= export SELECT_AI_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. FastAPI synchronous endpoints ============================= Create a file named ``app.py``: .. code-block:: python 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: .. code-block:: sh uvicorn app:app --host 0.0.0.0 --port 8000 Call the service: .. code-block:: sh 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. FastAPI asynchronous endpoints ============================== For async endpoints, initialize the async pool with ``select_ai.create_pool_async()`` and close it with ``select_ai.async_disconnect()``. .. code-block:: python 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: .. code-block:: sh uvicorn app:app --host 0.0.0.0 --port 8000 Press ``Ctrl+C`` to stop it. Flask example ============= Flask applications normally use synchronous request handlers, so initialize a synchronous Select AI pool when the application is created. .. code-block:: python 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 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. .. code-block:: python # 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")), ) .. code-block:: python # 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. 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: .. code-block:: text 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. 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. .. code-block:: python 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. 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.