semvec/pypi-packages huggingface dataset:
quillsql 3.0.0
Quill SDK for Python.

# Quill Python SDK

## Quickstart

First, install the quillsql package by running:

```bash
$ 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:

```python
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 = await 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!

## Streaming

```python
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")
```