wizsec

A Python SDK for the Wiz Cloud Security GraphQL API. Provides sync and async clients with automatic pagination, rate limiting, batch operations, and report generation.

Features

• Unified HTTP transport via httpx (sync and async)
• Automatic cursor-based pagination with result merging
• Per-environment rate limiting using pyrate-limiter (respects Wiz's global rate limits)
• Batch requests — submit multiple queries concurrently (sync threads or async tasks)
• Report generation — create, poll, stream, and download Wiz reports (JSON and CSV)
• Multiple auth flows — client credentials and device code (OAuth)
• Flexible credential storage — environment variables, credential files, or interactive prompt
• Multi-environment / multi-profile — connect to enabled app, gov, or fedramp environments with isolated credential profiles
• Serverless support — optimized for AWS Lambda and similar environments
• YAML configuration via ~/.wiz/wiz.config
• Client-side schema validation — catch query typos before they hit the API
• Custom query libraries — resolve query names from importable Python modules
• PEP 561 typed (py.typed marker included)

Installation

Install from Pypi:

pip install wizsec

Install from source:

git clone https://github.com/HusteDev/wizsec.git
cd wizsec
pip install .

For development:

pip install -e ".[dev]"

Requirements

• Python >= 3.9
• httpx, pyrate-limiter, python-dotenv, PyYAML, graphql-core

Quick Start

Authentication Setup

The SDK supports two OAuth grant types:

Client Credentials (default)

Provide your client ID and secret via environment variables, a credentials file, or constructor arguments.

Environment variables (simplest):

export WIZ_CLIENT_ID="your-client-id"
export WIZ_CLIENT_SECRET="your-client-secret"

Credentials file at ~/.wiz/wiz.credentials:

[default]
client_id = your-client-id
client_secret = your-client-secret
environment = app

Pass directly:

from wizsec import WizClient, Config

Config.load()
client = WizClient(client_id="...", client_secret="...")

Device Code (Interactive)

Device code auth opens a browser for the user to authorize the session — no client secret needed. This is ideal for CLI tools, notebooks, or any context where a human is present.

Set the grant type in ~/.wiz/wiz.config:

auth:
  grant_type: device_code
  device:
    quiet: true       # auto-authorize without extra prompts (default: true)
    poll_time: 5       # seconds between auth status checks (default: 5)

Then use the client normally — the browser will open automatically:

from wizsec import WizClient, Config

Config.load()
client = WizClient(environment="app")  # opens browser for authorization
result = client.create_request(query="...", vars={}).submit()

The SDK polls the auth endpoint until the user completes authorization or the request times out.

Your First Query

from wizsec import WizClient, Config

Config.load()
client = WizClient(environment="app")

response = client.create_request(
    query='{ users(first: 10) { nodes { name email } pageInfo { hasNextPage endCursor } } }',
    vars={}
)
result = response.submit()

if result.success:
    print(result.data)
else:
    print(result.errors)

Usage

Single Queries

response = client.create_request(query="...", vars={"first": 100})
result = response.submit()

Pagination is handled automatically — the SDK detects queries using the Relay connection pattern (nodes + pageInfo { hasNextPage endCursor }) and injects the $after cursor variable for you. Results from all pages are merged into result.data.

You don't need to declare $after in your query — the SDK adds it when:

• The operation is a query (not a mutation)
• The query selects both nodes and pageInfo subfields
• $after isn't already declared

Pagination defaults to api.auto_paginate in ~/.wiz/wiz.config. If you set paginate=False, no injection occurs and only the first page is returned.

Query Collections

Organize reusable GraphQL queries in a Python module and reference them by name. This keeps queries out of your application logic and makes them shareable across scripts.

queries.py — define queries as module-level constants:

ListUsers = """
    query ListUsers($first: Int) {
        users(first: $first) {
            nodes { id name email role }
            pageInfo { hasNextPage endCursor }
        }
    }
"""

GetProject = """
    query GetProject($id: ID!) {
        project(id: $id) { id name slug riskProfile { businessImpact } }
    }
"""

main.py — resolve by name or pass the string directly:

import queries

# Resolve by attribute name from the collection
response = client.create_request(
    queryCollection=queries,
    query="ListUsers",  # resolves to queries.ListUsers
    vars={"first": 50}
)

# Or pass the query string directly (no collection needed)
response = client.create_request(
    query=queries.GetProject,
    vars={"id": "some-project-id"},
    paginate=False,
)

queryCollection accepts a module, a module name string (auto-imported), or any object whose attributes are the query strings — for example a types.SimpleNamespace for one-off scripts that don't warrant a separate queries.py:

from types import SimpleNamespace

queries = SimpleNamespace(
    ListUsers="query ListUsers($first: Int) { users(first: $first) { nodes { id } } }",
)

response = client.create_request(queryCollection=queries, query="ListUsers", vars={"first": 50})

See examples/query_collection/ for a complete working example.

Batch Requests (Sync)

batch = client.create_batch_request()
batch.add_request(query="...", vars={"type": "VM"})
batch.add_request(query="...", vars={"type": "CONTAINER"})

batch.set_progress_callback(lambda done, total: print(f"{done}/{total}"))
results = batch.submit(max_concurrent=5)

print(f"{results.success_count()}/{results.total_count()} succeeded")

for request_id, response in results:
    if response.success:
        print(response.data)

Async Requests

import asyncio
from wizsec import WizClient, Config

async def main():
    Config.load()
    client = WizClient(environment="app")

    async with client.async_session() as async_client:
        response = await async_client.create_async_request(
            query="...", vars={"first": 100}
        )
        result = await response.submit()
        print(result.data)

asyncio.run(main())

Async Batch Requests

async with client.async_session() as async_client:
    batch = await async_client.create_async_batch_request()
    batch.add_request(query="...", vars={"type": "VM"})
    batch.add_request(query="...", vars={"type": "CONTAINER"})

    results = await batch.submit(max_concurrent=50)
    print(results.success_rate())

Async batch responses preserve one result per submitted request. Failed tasks remain indexed in the batch response with their errors attached.

Sync vs Async: When to Use Each

The SDK provides both synchronous and asynchronous interfaces. Choose based on your use case:

Performance comparison (3 queries fetching Projects, Users, and Service Accounts):

SYNC (sequential)    2.57s   — queries run one after another
ASYNC (concurrent)   0.60s   — queries run in parallel

Async achieves ~4x speedup here because all three API calls happen concurrently instead of waiting for each to complete.

Use sync when:

• Running a single query or a few dependent queries
• Writing simple scripts or one-off tools
• Code simplicity matters more than throughput

Use async when:

• Fetching data from multiple independent queries
• Building web applications or services that need to handle concurrent requests
• Performance is critical and queries don't depend on each other
• Working with asyncio-based frameworks (FastAPI, aiohttp, etc.)

See examples/sync_vs_async.py for a runnable comparison.

Report Generation

response = client.create_request(
    query="mutation { createReport(...) { report { id } } }",
    report_request={"name": "my-report", "stream": True}
)
result = response.submit()

# Report data is automatically polled, downloaded, and attached:
report_rows = result.data.get("report_data", [])

Progress Tracking

Monitor pagination progress with a callback that fires after each page is fetched. The on_page_event callback receives a dict with:

Simple progress logging:

def on_page(event):
    page = event["page_info"]["page"]
    per_page = event["page_info"]["per_page"]
    key = next(iter(event["page_data"]), None)
    count = len(event["page_data"][key]["nodes"]) if key else 0
    print(f"Page {page}: received {count}/{per_page} items")

response = client.create_request(
    query="...",
    vars={"first": 500},
    on_page_event=on_page
)
result = response.submit()

Spinner with live counter (runs the query on the background thread while animating in the main thread):

import sys, time, threading

progress = {"pages": 0, "items": 0, "done": False}

def on_page(event):
    progress["pages"] = event["page_info"]["page"]
    key = next(iter(event["page_data"]), None)
    if key:
        progress["items"] += len(event["page_data"][key].get("nodes", []))

result_holder = {}

def run_query():
    result_holder["result"] = client.create_request(
        query="...", vars={"first": 100}, on_page_event=on_page
    ).submit()
    progress["done"] = True

thread = threading.Thread(target=run_query)
thread.start()

spinner = "|/-\\"
i = 0
while not progress["done"]:
    sys.stdout.write(f"\r  {spinner[i % 4]} page {progress['pages']}, {progress['items']} items")
    sys.stdout.flush()
    i += 1
    time.sleep(0.15)
thread.join()

Progress tracking also works with async requests and report streaming. For reports, the callback receives {"name", "total_size", "downloaded", "status"} instead of page data.

See examples/progress_tracking.py for complete sync, spinner, and async examples.

Schema Validation

Validate GraphQL queries against the Wiz schema before they hit the API. Catches typos and invalid fields early with helpful suggestions:

from wizsec import WizClient, Config, SchemaValidator, WizSchemaValidationError

Config.load()
Config.set("api", "validate_queries", value=True)  # or set in wiz.config

client = WizClient()

# Typos are caught before the request is sent
try:
    client.create_request(query="query { projectz { nodes { id } } }")
except WizSchemaValidationError as e:
    print(e.validation_errors[0])
    # "Cannot query field 'projectz' on type 'Query'. Did you mean 'project', 'projects', or 'projectTags'?"

Validate queries programmatically without creating a request:

try:
    SchemaValidator.validate_query("query { fakeEndpoint { data } }", "app")
except WizSchemaValidationError as e:
    print(e.validation_errors[0])
    # "Cannot query field 'fakeEndpoint' on type 'Query'. Did you mean 'apiEndpoint'?"

The schema is cached under the configured app.wiz_dir as schema_<env>.json and reloaded automatically.

See examples/schema_validation.py for more examples.

Streaming Pagination

Auto-pagination aggregates every page in memory before returning. For large result sets, iterate instead — iterate_nodes fetches pages lazily and never holds more than one page:

for issue in client.iterate_nodes(query=ISSUES_QUERY, vars={"first": 500}):
    process(issue)          # breaking out early stops further API calls

async with client.async_session() as ac:
    async for issue in ac.iterate_nodes_async(query=ISSUES_QUERY, page_size=500):
        process(issue)

Each page goes through the normal rate-limiting and retry pipeline; a failed page raises the typed error (WizAPIError, WizRateLimitError, …).

Stopping Early

Sometimes you want a paginated query to stop as soon as the results cross a boundary — the first issue whose name matches something, or the point where a date-sorted result set passes your cutoff. Two optional predicates end pagination without fetching the rest:

from datetime import datetime, timezone

cutoff = datetime(2026, 1, 1, tzinfo=timezone.utc)

# Stop at the first issue older than the cutoff (results sorted newest first)
response = client.create_request(
    query=ISSUES_QUERY,
    vars={"first": 500},
    stop_when=lambda n: datetime.fromisoformat(n["createdAt"]) < cutoff,
)
result = response.submit()

print(len(result.data["issues"]["nodes"]), "issues fetched")
print(response.stopped_early, response.stop_info)

Given a page of [a, b, MATCH, d, e], the aggregated node list ends at MATCH — d and e are dropped along with every later page.

Stopping early is not an error. response.success stays True, and no exception is raised. Two properties tell you what happened:

• response.stopped_early — bool.
• response.stop_info — None if the query ran to completion, otherwise {"reason": "node" | "page", "page": int, "cursor": str | None, "next_cursor": str | None}.

Both predicates work everywhere pagination does — create_request, create_async_request, batch requests (each request stops itself), and iterate_nodes / iterate_nodes_async:

for issue in client.iterate_nodes(
    query=ISSUES_QUERY,
    vars={"first": 500},
    stop_when=lambda n: n["entitySnapshot"]["name"].startswith("legacy-"),
):
    process(issue)   # the matching issue is the last one yielded

A few things worth knowing:

• Resuming. stop_info["next_cursor"] is an exact resume point only when reason == "page". After a stop_when truncation it would skip the nodes you deliberately dropped, so resume from stop_info["cursor"] instead — that refetches the whole stopping page, and you de-duplicate by id. Relay nodes selections carry no per-node cursor, so there is no exact mid-page resume point.
• totalCount is untouched. It stays the server-side total, so it will exceed the number of nodes you got back. stopped_early is the signal that the list is partial.
• Predicates run inline on the fetch loop (the queue worker thread for sync requests), so keep them cheap and thread-safe. A predicate that raises is reported as a WizQueryError — success becomes False — rather than silently returning results truncated at an arbitrary point.
• Query splitting is disabled for any request with a predicate set. Split sub-queries run concurrently over disjoint scopes, so "the first matching node" would have no global ordering.

See examples/early_stop.py for complete sync, async, iterator, and resume examples.

Rate Limiting

The SDK automatically enforces Wiz's API rate limits so you don't have to think about throttling. Rate limiters are shared across all WizClient instances on the same environment, even across different profiles.

Limits are applied per request type (query vs. mutation) and account type (user vs. service account), based on Wiz's published rate limits.

By default the SDK runs at 80% of the published limits. The headroom keeps normal operation clear of server-side 429s — the server's quota is per tenant, so network jitter and other consumers (other scripts, CI jobs, integrations) draw from the same budget. Both the headroom and the absolute rates are configurable:

rate_limit:
  headroom: 0.8            # fraction of published limits to use (default 0.8)
  overrides:               # optional absolute requests-per-second per limiter key
    query_service: 8       # wins over headroom for that key

Local limiter waits and server 429 backoffs are handled automatically and never consume retry attempts. Rate-limit responses reported inside a 200 response's GraphQL errors list are detected and backed off the same way. Async clients on the same environment share the same server backoff window.

Note that the local limiter coordinates requests within a single Python process. Separate processes each apply their own budget, so if you run several wizsec processes in parallel against one tenant, lower the headroom (or set explicit overrides) accordingly.

CLI

Installing the package registers a wizsec command for managing configuration and credentials:

wizsec config init            # create a default config file (~/.wiz/wiz.config)
wizsec config show            # print the current config
wizsec config get api.timeout
wizsec config set rate_limit.headroom 0.7
wizsec config unset query_splitting.enabled

wizsec creds set --profile default --environment gov   # prompts for ID/secret
wizsec creds list             # profiles with masked IDs (secrets never shown)
wizsec creds remove staging
wizsec creds test             # authenticate with stored credentials

wizsec schema refresh         # re-introspect the API and rewrite the cached schema
wizsec schema path            # print the schema cache path
wizsec schema clear           # delete cached schemas (--environment for just one)

wizsec doctor                 # diagnose config, credentials, rate budgets, schema caches
wizsec doctor --auth          # same, plus a live authentication check

All commands accept --file to operate on a non-default location. config set rewrites the YAML file, which drops comments — see the bundled template for documentation of every option.

Keeping the schema current

Query validation uses a schema cached at ~/.wiz/schema_<env>.json. It is fetched automatically the first time it is needed, but only when the file is absent — nothing refetches it as it ages. wizsec doctor warns once a cache passes 30 days, and wizsec schema refresh is how you act on that warning.

wizsec schema refresh authenticates like creds test, so it accepts --profile and --environment.

For serverless deployments the bundle is read-only and cannot introspect at runtime, so generate the schema ahead of time and ship it alongside your code:

wizsec schema refresh --environment app --output ./bundle/schema_app.json

With --output the schema is written only to that path — your own ~/.wiz cache is left untouched.

Configuration

The SDK reads ~/.wiz/wiz.config (YAML). Example:

app:
  name: wizsec
  release: "1.1.0"
  config_schema: 2

auth:
  grant_type: client_credentials
  credentials:
    storage_method: file
    file_path: ~/.wiz/
  proxy:
    http:
      url: ""
      port: 80
    https:
      url: ""
      port: 80

domain:
  default: gov
  app:
    enabled: false
  gov:
    enabled: true
  fedramp:
    enabled: false

api:
  timeout: 60
  max_retries: 3
  retry_time: 2
  auto_paginate: true
  validate_queries: false

logging:
  enabled: false
  console_handler:
    enabled: true
    logging_level: INFO

Blank proxy URLs use environment proxy variables. Config can also be set via Config.load(overrides=["api.timeout=120"]).

Multi-Environment & Multi-Profile

# Different Wiz tenants
app_client = WizClient(environment="app")
gov_client = WizClient(environment="gov")

# Different credential profiles on the same tenant
admin = WizClient(environment="app", profile="admin")
readonly = WizClient(environment="app", profile="readonly")

Clients sharing the same environment automatically share a single request queue and rate limiter. Requested environments must be enabled under domain.<environment>.enabled. Auth state is isolated by (environment, profile), so the same profile name can be used safely against different environments.

Serverless / Lambda

Set WIZ_SERVERLESS=1 or deploy to an environment with AWS_LAMBDA_FUNCTION_NAME set. The SDK adapts automatically:

• Disables background worker threads (executes inline)
• Reads config from /var/task/.wiz/
• Call client.cleanup_for_lambda() at the end of each invocation

def handler(event, context):
    Config.load()
    client = WizClient(environment="app", serverless=True)
    try:
        result = client.create_request(query="...", vars={}).submit()
        return result.data
    finally:
        client.cleanup_for_lambda()

Error Handling

The SDK provides a structured exception hierarchy:

WizReportError is deprecated and no longer exported from wizsec. Nothing raises it — report failures surface as entries in response.errors and, where a typed error is set, as WizTimeoutError or WizAPIError on response.error. It remains importable from wizsec.exceptions so existing except clauses keep working, and will be removed in a future major release.

submit() never raises for API-level failures — inspect the response instead. Failed requests carry a typed exception on response.error (WizRateLimitError when the server rate limit won, WizAPIError with status_code otherwise), and response.raise_on_error() converts a failure into that exception:

from wizsec import WizRateLimitError

response = client.create_request(query=QUERY, vars={"first": 100})
result = response.submit()

if not response.success:
    if isinstance(response.error, WizRateLimitError):
        print(f"Rate limited — retry after {response.error.retry_after}s")
    else:
        print(f"Failed: {response.errors}")

# or, exception-style:
response.raise_on_error()

iterate_nodes raises the typed error directly, since an iterator has no response object to inspect.

Development

pip install -e ".[dev]"     # install with dev + docs dependencies
python -m pytest tests/ -q  # run tests

Documentation

API docs are built with MkDocs Material:

pip install -e ".[docs]"
mkdocs serve                # live preview at http://127.0.0.1:8000
mkdocs build                # static site in site/

License

MIT