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 = None
@app.on_event("startup")
async def startup_event():
global quill
quill = await Quill.create(
private_key=os.getenv("QULL_PRIVATE_KEY"),
database_connection_string=os.getenv("POSTGRES_READ"),
database_type="postgresql"
)
@app.on_event("shutdown")
async def shutdown_event():
if quill is not None:
await quill.aclose()
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
from quillsql import Quill
from fastapi.responses import StreamingResponse
import asyncio
quill = None
@app.on_event("startup")
async def startup_event():
global quill
quill = await Quill.create(
private_key=os.getenv("QULL_PRIVATE_KEY"),
database_connection_string=os.getenv("POSTGRES_READ"),
database_type="postgresql"
)
@app.on_event("shutdown")
async def shutdown_event():
if quill is not None:
await quill.aclose()
@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")
Metadata
Release files for quillsql 3.0.3
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| quillsql-3.0.3.tar.gz | 31.8 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| quillsql-3.0.3-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 64.5 kB
Release files / quillsql-3.0.3.tar.gz
| Download URL | quillsql-3.0.3.tar.gz |
|---|---|
| Size | 31.8 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
837df2bea0b1a03881133965581d3ab46056bda70c57cbed87b3f3b1104f81bf
|
|
BLAKE2b-256 checksum How to use checksums |
06fb12a0b1c6ac40e0aa1e41dc49d70511b28be091c129cf82a804b9ee7f2651
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.2.0 CPython/3.12.8
|
Release files / quillsql-3.0.3-py3-none-any.whl
| Download URL | quillsql-3.0.3-py3-none-any.whl |
|---|---|
| Size | 32.6 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
7dae7c9bfb0249f8bdb457a6300781007cb29fbdde45e4574098697daabcc490
|
|
BLAKE2b-256 checksum How to use checksums |
c73bfd458225cce1fdb58135e9fc583b1b8f92d12cf53268941e6f294e08257c
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.2.0 CPython/3.12.8
|