squeezle (Python client)

A small, dependency-light Python client for the Squeezle SQL query API. Authenticate with a personal sqz_ token, run saved queries or ad-hoc SQL, and get the rows back.

Install

pip install squeezle            # from this directory: pip install -e .
pip install "squeezle[pandas]"  # add .to_pandas() support

Only runtime dependency: requests.

Quick start

from squeezle import Client

sqz = Client("sqz_...")                    # or: Client.from_env()

# Run a saved query and block until it finishes
run = sqz.run_saved("query-uuid", variables={"since": "2026-01-01"})

for row in run.dicts():                    # up to ~1000 rows (stored preview)
    print(row)                             # {"id": 7, "email": "a@b.com"}

df = run.to_pandas()                       # needs the [pandas] extra

Client.from_env() reads SQUEEZLE_TOKEN and, optionally, SQUEEZLE_BASE_URL (default https://api.squeezle.app).

Ad-hoc SQL

run = sqz.run_sql(
    data_source_id="ds-uuid",
    sql="SELECT status, count(*) FROM orders WHERE created_at > {{since}} GROUP BY 1",
    variables={"since": "2026-01-01"},
    row_limit=5000,                        # int, or "max"
)
print(run.dicts())

Your token needs the queries:write scope to run ad-hoc SQL, plus runs:read to read results.

Variables

{{name}} placeholders in SQL bind as typed Postgres params ($1, $2, ...), so values can never change the query shape (injection-safe). For a saved query the types are stored, so you pass only variables. For ad-hoc SQL, declare each variable's type in variable_definitions.

Three things that trip people up:

1. Lists use = ANY(...), not IN. A *[] variable binds as one array param, so write WHERE id = ANY({{ids}}). WHERE id IN {{ids}} is invalid SQL.
2. A range is one variable, two placeholders. Define period as date_range, then write {{period_start}} and {{period_end}} in the SQL; pass {"period": {"start": ..., "end": ...}}.
3. required defaults to True. For an optional variable set "required": False and a "default", or you get a 422 "is required".

Enum / dropdown is type: "text" with control: "select" and options. Datetime is type: "timestamp". There is no number/list/enum type name.

run = sqz.run_sql(
    "ds-uuid",
    "SELECT * FROM orders WHERE status = ANY({{statuses}}) AND created_at >= {{since}}",
    variables={"statuses": ["paid", "shipped"], "since": "2026-01-01"},
    variable_definitions=[
        {"name": "statuses", "type": "text[]"},
        {"name": "since", "type": "date"},
    ],
)

See examples/variables.py for a runnable example of every type.

Getting all the data (beyond 1000 rows)

The API keeps only a ~1000-row preview in the database. For the full result, export to an artifact and download it:

rows = run.fetch_all("json")               # list[dict], the whole result
csv_text = run.fetch_all("csv")            # str
run.download("xlsx", "orders.xlsx")        # straight to disk (csv/json/jsonl/xlsx)
url = run.export_url("csv")                # short-lived signed URL, do it yourself

Two-step control (start now, wait later)

run = sqz.start_saved("query-uuid", variables={"n": 3})   # returns immediately (queued)
run.wait(timeout=120, poll_interval=1.0)                  # poll until terminal
run.raise_for_status()                                    # raise unless it succeeded

Manage saved queries

q = sqz.create_query(
    name="Daily orders",
    sql="SELECT * FROM orders WHERE created_at > {{since}}",
    data_source_id="ds-uuid",
    folder_id="folder-uuid",          # optional; visibility, tags, ... too
)

# Edit. The API optimistic-locks on `version`; omit it and the client reads the
# current version first (pass one you already hold to avoid the extra GET).
sqz.update_query(q["id"], sql="SELECT * FROM orders LIMIT 100")
sqz.update_query(q["id"], version=q["version"], name="Renamed")

sqz.delete_query(q["id"])

A stale version raises ConflictError (details["current_version"]). Query writes need the queries:write scope.

Browse the workspace

sqz.me()                     # user, capabilities, active org, plan
sqz.list_data_sources()      # connections you can query
sqz.list_queries()           # saved queries
sqz.get_query("query-uuid")
sqz.list_runs(query_id="query-uuid", limit=10)
sqz.get_run("run-uuid")
sqz.compile("SELECT {{id}}") # discover a query's variables without running it

Errors

Every API error raises a subclass of SqueezleError carrying .code, .message, .details, and .status:

from squeezle import ValidationError

try:
    sqz.run_sql("ds-uuid", "SELECT bad")
except ValidationError as exc:
    print(exc.code, exc.details)

Test

python -m unittest discover -s tests    # stdlib only, no network
# or: pytest