Python SDK

The Querri Python SDK (querri) gives you synchronous and asynchronous clients for embedding Querri analytics, managing users and access policies, reading and writing data, and controlling embed sessions from Python.

Requires Python 3.10 or later. Built on httpx and Pydantic v2, and fully typed.

Installation

The package comes in two forms:

pip install querri          # the SDK
pip install "querri[cli]"   # the SDK plus the querri command-line tool

For a server-side application, the plain install is enough. The cli extra is only for the querri command; see the CLI page.

Quick start

import os
from querri import Querri

client = Querri(
    api_key=os.environ["QUERRI_API_KEY"],
    org_id=os.environ["QUERRI_ORG_ID"],
)

for project in client.projects.list():
    print(project.name)

client.close()

Or let the SDK read the environment:

export QUERRI_API_KEY="qk_live_..."
export QUERRI_ORG_ID="your_org_id"

from querri import Querri

client = Querri()  # reads QUERRI_API_KEY and QUERRI_ORG_ID

Authentication

The client looks for credentials in this order:

1. An API key: api_key= or QUERRI_API_KEY. Keys start with qk_, and an admin creates them in Settings → API Keys.
2. An access token: access_token= or QUERRI_ACCESS_TOKEN.
3. Your saved sign-in from querri auth login.

With an API key you also need the organization ID (org_id= or QUERRI_ORG_ID). With a token, the organization comes from the token. Explicit arguments always beat environment variables. If both a key and a token are set, the token is sent.

Configuration options

Parameter	Environment variable	Default	Description
api_key	QUERRI_API_KEY	none	Your qk_ API key
access_token	QUERRI_ACCESS_TOKEN	none	An access token, used instead of a key
org_id	QUERRI_ORG_ID	none	Organization ID, required with an API key
host	QUERRI_HOST	https://app.querri.com	Server host
timeout	QUERRI_TIMEOUT	30.0	Request timeout in seconds
max_retries	QUERRI_MAX_RETRIES	3	Retries for failed requests
profile		none	Which saved sign-in profile to use

Note

The parameter is host, not base_url. The SDK adds /api/v1 for you. For local development, use Querri(host="http://localhost").

client = Querri(
    api_key="qk_live_...",
    org_id="your_org_id",
    host="http://localhost",  # local development
    timeout=60.0,
    max_retries=5,
)

get_session: the main entry point

client.embed.get_session() is the method most embeds need. It finds or creates the user, applies their access, and creates an embed session, all in one call.

Wire a session endpoint

Your backend calls get_session() for the signed-in user and hands the token to the browser. Always pass on the request’s Origin header. If your organization has a list of allowed embed domains, a missing origin fails with 400 origin_required, and one that isn’t on the list fails with 403 origin_not_allowed.

session = client.embed.get_session(
    user={"external_id": "customer-42", "email": "alice@customer.com"},
    origin=request.headers.get("origin"),
    ttl=3600,
)
print(session["session_token"])

Dict form: find or create the user

session = client.embed.get_session(
    user={
        "external_id": "customer-42",
        "email": "alice@customer.com",
        "first_name": "Alice",
        "last_name": "Smith",
        "role": "member",
    },
    origin="https://app.customer.com",
)

external_id is required. Include email so the user can be created if they don’t exist yet.

String shorthand: existing users only

session = client.embed.get_session(
    user="customer-42",
    origin="https://app.customer.com",
)

A string looks up an existing user by external ID. If there’s no such user, it raises ValueError; pass a dict to create users automatically.

Inline access: managed policies

Pass sources and filters, and the SDK creates a policy for that exact access, or reuses one that already matches:

session = client.embed.get_session(
    user={"external_id": "customer-42", "email": "alice@customer.com"},
    access={
        "sources": ["8f14e45f-ceea-4ba1-9f3c-5d2b8a0c1e77"],
        "filters": {
            "region": ["APAC", "EMEA"],
            "department": "Sales",
        },
    },
    origin="https://app.customer.com",
    ttl=3600,
)

Existing policies

session = client.embed.get_session(
    user="customer-42",
    access={"policy_ids": ["<policy_uuid>", "<policy_uuid>"]},
    origin="https://app.customer.com",
)

Return value

get_session() returns a plain dict:

{
    "session_token": "es_...",
    "expires_in": 3600,
    "user_id": "...",
    "external_id": "customer-42",
}

ttl is in seconds, from 900 to 86400 (default 3600). A value outside that range raises ValueError before any request is sent.

How managed policies work

With inline access, the SDK hashes the sources and filters into a policy name such as sdk_auto_a1b2c3d4. Users with identical access share that policy. Each call with an access block replaces all of the user’s policy assignments with the new set, so filters from an earlier call don’t linger.

Embed UI settings

client.embed.get_ui_config(org) returns the embed settings configured for an organization (chrome, theme and privacy). It calls a public endpoint that needs no key.

Access policies

Access policies control which rows of a source a user sees.

Create a policy

policy = client.policies.create(
    name="APAC Sales",
    description="Restricts data to the APAC region",
    source_ids=["<source_uuid>", "<source_uuid>"],
    row_filters=[
        {"column": "region", "values": ["APAC"]},
        {"column": "department", "values": ["Sales", "Marketing"]},
    ],
)

Assign users

client.policies.assign_users(policy.id, user_ids=[user.id])

The setup() shortcut

Create a policy and assign users in one call, with row filters as a dict:

policy = client.policies.setup(
    name="APAC Sales Team",
    sources=["<source_uuid>", "<source_uuid>"],
    row_filters={"region": ["APAC"], "department": "Sales"},
    users=["<user_id>", "<user_id>"],
)

Other operations

# List policies, or find them by name
policies = client.policies.list()
policies = client.policies.list(name="APAC Sales")

# Get, update, delete
policy = client.policies.get(policy_id)
client.policies.update(policy_id, name="New Name")
client.policies.delete(policy_id)

# Remove one user, or replace all of a user's policies
client.policies.remove_user(policy_id, user_id)
client.policies.replace_user_policies(user_id, policy_ids=[policy_id])

# Filterable columns for a source
columns = client.policies.columns(source_id=source_id)

# The filters a user ends up with on a source
resolved = client.policies.resolve(user_id=user_id, source_id=source_id)

How policies combine

• Same column, several policies: OR. Policies for region = US and region = EU mean the user sees US or EU rows.
• Different columns: AND. Policies for region = US and department = Sales mean only rows that match both.
• No policies: every row, except in sources marked access-controlled, which show no rows until a policy applies.

User management

# Create a user
user = client.users.create(
    email="alice@example.com",
    external_id="cust-42",
    first_name="Alice",
    last_name="Smith",
    role="member",  # "member" or "admin"
)
print(user.id, user.email)

# Get a user by ID
user = client.users.get(user_id)

# Get or create by external ID
user = client.users.get_or_create(
    external_id="cust-42",
    email="alice@example.com",
    first_name="Alice",
)

# List users (loops through every page)
for user in client.users.list():
    print(user.email)

# Filter by external ID
page = client.users.list(external_id="cust-42")
user = page.data[0]

# Update and delete
updated = client.users.update(user.id, first_name="Alicia")
client.users.delete(user.id)

# Unlink an external ID without deleting the user
client.users.remove_external_id("cust-42")

role="member" shows as Creator on the People page in Querri.

Data sources

Reading and writing data lives on client.sources. The Data API page has fuller examples.

Create a source from rows

source = client.sources.create_data_source(
    name="Sales Data",
    rows=[
        {"region": "US", "revenue": 100000},
        {"region": "EU", "revenue": 85000},
    ],
)
print(source.id, source.row_count)

List, get, update, delete

for s in client.sources.list():
    print(s["id"], s["name"])

source = client.sources.get(source_id)
client.sources.update(source_id, name="Renamed Source")
client.sources.delete(source_id)

Query and read

# SQL against one source, with row-level security applied. The source is
# always a view named `data`: write FROM data, never the display name.
result = client.sources.query(
    sql="SELECT region, SUM(revenue) AS total FROM data GROUP BY region",
    source_id=source_id,
    page=1,
    page_size=100,
)
print(result.data)
print(result.total_rows)

# Rows, a page at a time
page = client.sources.source_data(source_id, page=1, page_size=100)

# A question: Querri writes and runs the SQL
answer = client.sources.ask(source_id, question="How many rows do we have?")
print(answer["generated_sql"], answer["data"])

ask takes the question as a keyword argument.

Change the data

# Add rows
client.sources.append_rows(source_id, rows=[{"region": "APAC", "revenue": 60000}])

# Replace every row
client.sources.replace_data(source_id, rows=fresh_rows)

client.sources.sync() calls an endpoint that isn’t implemented yet, and client.sources.create() (with a connector) has no matching endpoint. Neither works today.

Embed sessions

For lower-level control than get_session():

# Create a session
session = client.embed.create_session(
    user_id=user_id,
    origin="https://app.customer.com",
    ttl=3600,  # seconds, 900 to 86400
)
print(session.session_token)  # "es_..."

# Swap for a new token (the old one is revoked)
new_session = client.embed.refresh_session(session_token=session.session_token)

# List active sessions
session_list = client.embed.list_sessions(limit=50)
for s in session_list.data:
    print(s.session_token, s.user_id)

# Revoke one session, or every session for a user
client.embed.revoke_session(session_token=new_session.session_token)
revoked = client.embed.revoke_user_sessions(user_id)

User-scoped client

client.as_user(session) returns a client that acts as the session’s user, so results are limited to what that user can access:

session = client.embed.get_session(user="customer-42", origin="https://app.customer.com")
user_client = client.as_user(session)

for project in user_client.projects.list():
    print(project.name)

It has projects, dashboards, sources and chats. Dashboards are read-only for a session.

Projects and dashboards

# List projects (loops through every page)
for project in client.projects.list():
    print(project.name)

# Create a project
project = client.projects.create(
    name="Q4 Analysis",
    user_id=user_id,
    description="Quarterly sales analysis",
)

# Run it and check on it
run = client.projects.run(project.id, user_id=user_id)
status = client.projects.run_status(project.id)

# Steps and their rows (row-level security applied)
steps = client.projects.list_steps(project.id)
data = client.projects.get_step_data(project.id, steps[0].id, page=1, page_size=100)

# Dashboards
for dashboard in client.dashboards.list():
    print(dashboard.name)
client.dashboards.refresh(dashboard_id)

client.dashboards.create() and client.dashboards.delete() call endpoints that aren’t implemented yet.

Chat streaming

Stream a response in a project chat:

chat = client.projects.chats.create(project.id)

stream = client.projects.chats.stream(
    project.id,
    chat.id,
    prompt="Summarize the sales data by region",
    user_id=user_id,
)
for chunk in stream:
    print(chunk, end="", flush=True)

To wait for the whole response instead, call stream.text() on a fresh stream. stream.events() yields structured events if you need more than text.

Async client

AsyncQuerri mirrors the sync client with async and await:

import asyncio
from querri import AsyncQuerri

async def main():
    client = AsyncQuerri()

    # Loops through every page
    async for project in client.projects.list():
        print(project.name)

    # Streaming
    stream = await client.projects.chats.stream(
        project_id, chat_id,
        prompt="Summarize the data",
        user_id=user_id,
    )
    async for chunk in stream:
        print(chunk, end="", flush=True)

    # get_session works the same way
    session = await client.embed.get_session(
        user={"external_id": "cust-42", "email": "a@b.com"},
        access={"sources": ["<source_uuid>"]},
        origin="https://app.customer.com",
    )

    await client.close()

asyncio.run(main())

Async pagination

An async page has no .data property. Use await page.get_data() for the first page, or async for item in page to loop through every page.

Error handling

API errors inherit from APIError, which carries status, code, type, doc_url and request_id:

from querri import (
    APIError,
    AuthenticationError,
    NotFoundError,
    OriginRequiredError,
    RateLimitError,
    ValidationError,
    ServerError,
)

try:
    project = client.projects.get("nonexistent-id")
except NotFoundError as e:
    print(f"Not found (status={e.status}, code={e.code})")
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except AuthenticationError:
    print("Invalid API key or token")
except OriginRequiredError:
    print("This organization requires an origin for embed sessions")
except ValidationError as e:
    print(f"Bad request: {e}")
except ServerError as e:
    print(f"Server error: {e.status}")
except APIError as e:
    print(f"API error {e.status}: {e}")
    print(f"  type={e.type}, code={e.code}, request_id={e.request_id}")

Exception hierarchy

QuerriError
├── APIError                   any other status, such as 413 or 501
│   ├── AuthenticationError    401
│   ├── PermissionError        403
│   ├── NotFoundError          404
│   ├── ValidationError        400
│   │   └── OriginRequiredError   400 origin_required
│   ├── ConflictError          409
│   ├── RateLimitError         429, with retry_after
│   └── ServerError            500, 502, 503
├── StreamError
│   ├── StreamTimeoutError
│   └── StreamCancelledError
└── ConfigError

Retries

The SDK retries up to max_retries times (default 3), backing off between attempts and honoring Retry-After:

• 429 is always retried.
• 500, 502 and 503 are retried only for GET, PUT, DELETE, HEAD and OPTIONS. A POST or PATCH that fails that way raises straight away, so it isn’t sent twice.

Complete example: multi-tenant embed

A FastAPI route that creates a session for each tenant’s user:

import os

from fastapi import Depends, FastAPI, Request
from querri import AsyncQuerri

app = FastAPI()
querri = AsyncQuerri()  # reads from environment variables

@app.get("/api/querri-token")
async def get_querri_token(
    request: Request,
    current_user=Depends(get_current_user),  # your own auth dependency
):
    session = await querri.embed.get_session(
        user={
            "external_id": str(current_user.id),
            "email": current_user.email,
            "first_name": current_user.first_name,
            "last_name": current_user.last_name,
        },
        access={
            "sources": [os.environ["QUERRI_SOURCE_ID"]],
            "filters": {"tenant_id": str(current_user.tenant_id)},
        },
        origin=request.headers.get("origin"),
    )
    return {"sessionToken": session["session_token"]}

Each tenant sees only their own rows: one dataset, one embed, filtered per user. Pair it with the Embed SDK on the frontend.

Upgrading to 2.0.0

The main changes from 1.x:

• Embed sessions no longer take source_scope. Control access with access and policies.
• OriginRequiredError is raised when your organization needs an origin and none was sent.
• ttl is checked before sending and must be 900 to 86400.
• New: client.embed.get_ui_config().

The 1.x line gets security fixes only, for six months from the 2.0.0 release on 2026-08-31.

Security best practices

• Keep API keys (qk_) on the server. Never put them in client code or version control.
• Session tokens (es_) are safe in the browser. They’re tied to one user and expire.
• Pass origin to get_session() so each session is bound to your site.
• Store credentials in environment variables.
• Replace API keys before they expire, from Settings → API Keys.

API reference

The Python SDK wraps the Querri public API. For every endpoint, see the API Reference.

SDK resource	API endpoints (under /api/v1)
client.users	/users, /users/external/{external_id}
client.embed	/embed/sessions
client.policies	/access/policies, /access/resolve, /access/columns, /access/users/{user_id}/policies
client.files	/files
client.projects	/projects, /projects/{project_id}/chats
client.dashboards	/dashboards
client.sharing	/projects/{project_id}/shares, /dashboards/{dashboard_id}/shares, /sources/{source_id}/shares, /sources/{source_id}/org-share
client.sources	/sources, /connectors
client.views	/views
client.keys	/keys
client.audit	/audit/events
client.usage	/usage