quillsql 2.2.11

Quill SDK for Python.

Quill Python SDK

Quickstart

First, install the quillsql package by running:

$ pip install quillsql

Then, add a /quill endpoint to your existing python server. For example, if you were running a FASTAPI app, you would just add the endpoint like this:

from quillsql import Quill

quill = Quill(
    private_key=os.getenv("QULL_PRIVATE_KEY"),
    database_connection_string=os.getenv("POSTGRES_READ"),
    database_type="postgresql"
)

security = HTTPBearer()

async def authenticate_jwt(token: str = Depends(security)):
    # Your JWT validation logic here
    # Return user object or raise HTTPException
    user = validate_jwt_token(token.credentials)
    return user

@app.post("/quill")
async def quill_post(data: Request, user: dict = Depends(authenticate_jwt)):
    # assuming user fetched via auth middleware has an userId
    user_id = user["user_id"]
    body = await data.json()
    metadata = body.get("metadata")

    result = quill.query(
        tenants=[{"tenantField": "user_id", "tenantIds": [user_id]}],
        metadata=metadata
    )
    return result

Then you can run your app like normally. Pass in this route to our react library on the frontend and you all set!

Performance Tuning And Profiling

You can tune Postgres concurrency and enable profiling with either cache config or environment variables.

quill = Quill(
    private_key=os.getenv("QULL_PRIVATE_KEY"),
    database_connection_string=os.getenv("POSTGRES_READ"),
    database_type="postgresql",
    cache={
        "postgres_pool_min": 4,
        "postgres_pool_max": 24,
        "run_query_max_workers": 8,
        "db_max_inflight": 12,
        "profile_logging": True,
    },
)

Supported environment variables:

• QUILL_POSTGRES_POOL_MIN
• QUILL_POSTGRES_POOL_MAX
• QUILL_RUN_QUERY_MAX_WORKERS
• QUILL_DB_MAX_INFLIGHT
• QUILL_PROFILE_LOGGING

Worker behavior is adaptive for heavy dashboard/report requests:

• report / pivot-template default to 1 worker.
• They step up to 2 workers only when recent pool-acquire wait is low.
• They scale to 4 only for high fanout when pool-acquire wait is very close to zero.
• They back off aggressively when pool-acquire wait rises.

When profiling is enabled, logs include a per-request request_id so you can correlate API call timing and DB execution timing for the same SDK query.

Streaming

from quillsql import Quill
from fastapi.responses import StreamingResponse
import asyncio

quill = Quill(
    private_key=os.getenv("QULL_PRIVATE_KEY"),
    database_connection_string=os.getenv("POSTGRES_READ"),
    database_type="postgresql"
)

@app.post("/quill-stream")
async def quill_post(data: Request, user: dict = Depends(authenticate_jwt)):
    # assuming user fetched via auth middleware has an userId
    user_id = user["user_id"]
    body = await data.json()
    metadata = body.get("metadata")

    quill_stream = quill.stream(
        tenants=[{"tenantField": "user_id", "tenantIds": [user_id]}],
        metadata=metadata,
    )

    async def event_generator():
        # Full event types list: https://ai-sdk.dev/docs/ai-sdk-ui/stream-protocol#data-stream-protocol
        async for event in quill_stream:
            if event["type"] == "start":
                pass
            elif event["type"] == "text-delta":
                yield event['delta']
            elif event["type"] == "finish":
                return
            elif event["type"] == "error":
                yield event['errorText']
            await asyncio.sleep(0)

    return StreamingResponse(event_generator(), media_type="text/event-stream")