snowloader

Get ServiceNow data into your AI pipeline without quietly losing rows.

Incidents, Knowledge Base, CMDB, Changes, Problems, Catalog and attachments,
into LangChain, LlamaIndex, or your own code. Every sweep sorts on a unique key,
so offset pagination cannot drop records, and it can prove it returned everything.

Documentation  ·  PyPI  ·  Install  ·  Upgrading  ·  API  ·  Roadmap

Every release before 0.3.0 paged over a column that is not unique. The count still reconciled, so nothing reported it.

Two ways in

From a shell, one command. Resumable, and it checks its own work:

pip install snowloader

export SNOW_INSTANCE=https://yourcompany.service-now.com
export SNOW_USER=api_user
export SNOW_PASS=...

snowloader extract incident --out incidents.jsonl \
    --query "stateIN6,7^close_notesISNOTEMPTY" \
    --fields sys_id,number,close_notes --display-value all --resume

Run it, kill it, run it again and it continues. It exits non-zero if the sweep did not return every record, so an unattended job finds out.

From Python, three lines. The same loader objects work with LangChain, LlamaIndex, or anything else that takes a list of dicts:

from snowloader import SnowConnection, IncidentLoader

with SnowConnection(instance_url="https://yourcompany.service-now.com",
                    username="api_user", password="api_pass") as conn:
    docs = IncidentLoader(connection=conn, query="active=true").load(verify=True)

The problem it solves

ServiceNow does not guarantee a stable row order inside a group of rows that tie on the sort column. Every release before 0.3.0 paged over sys_created_on, which is not unique, so a page boundary landing inside a tied group returned some rows twice and skipped others.

The returned count still matched, so nothing reported a problem. The loss was identical on every run, so re-running never revealed it. Measured on a developer instance where cmdb_ci holds 2,919 rows across only 818 distinct timestamps, three consecutive sweeps each returned 2,919 rows and only 2,915 of them were distinct.

Every ORDERBY chain ends in sys_id, so that boundary cannot fall inside a tie, and you can ask a sweep to prove it was complete:

records = list(conn.get_records("cmdb_ci", verify=True))   # raises if anything is missing

The second thing it fixes is that a document could not be joined to anything. The loaders returned labels and threw the identifiers away.

A long sweep can also be interrupted and continued. Ordering, verification and resume are the three things that were being assembled by hand and getting quietly wrong, so 0.5.0 made them the defaults of a command:

snowloader extract cmdb_ci --out cmdb.jsonl --resume --workers 16

Details in the upgrade notes.

Upgrading from 0.2.x

0.3.0 fixed a data loss bug. If you are still on 0.2.x, upgrade.

ServiceNow offset pagination is only safe over a unique sort key. The API does not guarantee a stable order inside a group of rows that tie on the sort column, so a page boundary landing inside a tied group returns some rows twice and skips others. Every release before 0.3.0 sorted on sys_created_on, which is not unique.

Measured on a developer instance, cmdb_ci holds 2,919 rows across only 818 distinct sys_created_on values, worst tie 31 rows. Three consecutive sweeps:

run 1: returned=2919 distinct=2915 lost=4 duplicated=4 count_matches_api=True
run 2: returned=2919 distinct=2915 lost=4 duplicated=4 count_matches_api=True
run 3: returned=2919 distinct=2915 lost=4 duplicated=4 count_matches_api=True

The count reconciles, so nothing reports a problem. The loss is deterministic, so re-running never surfaces it and diffing two runs shows nothing. On a 30,000 row CMDB that rate is roughly 40 missing CIs.

0.3.0 ends every ORDERBY chain with sys_id. Nothing in your code changes. The same table now returns all 2,919 distinct rows on both the sequential and the threaded path.

You can also ask a sweep to prove it was complete:

records = list(conn.get_records("cmdb_ci", verify=True))   # raises if anything is missing

Two breaking changes come with it, both in loader metadata. The one to check before you upgrade a retrieval pipeline: metadata now carries every field on the record plus an identifier beside each reference, which on a wide table is roughly three times the previous size. If you write the whole metadata dict into a vector store with a per-vector size limit, pass expand_references=False to keep the 0.2.x shape. Full detail in the changelog and the docs.

Architecture

snowloader sits between ServiceNow's Table API and whatever LLM stack you are building. The connection layer handles auth, pagination, retries, and rate limiting. The loaders normalize each table into a SnowDocument. The adapters translate that into LangChain Document or LlamaIndex Document types without copying business logic.

Why snowloader?

Building RAG or agentic AI on top of ServiceNow data. snowloader covers the core tables plus a generic loader for the rest, gives you sync, threaded and async paginators, and keeps the core free of any framework so you can plug it into LangChain, LlamaIndex, or your own pipeline.

Installation

# pip
pip install snowloader              # Core only
pip install snowloader[async]       # + AsyncSnowConnection (aiohttp)
pip install snowloader[langchain]   # + LangChain adapter
pip install snowloader[llamaindex]  # + LlamaIndex adapter
pip install snowloader[all]         # Everything

# uv
uv add snowloader
uv add snowloader[all]

Requirements: Python 3.10, 3.11, 3.12, or 3.13. A ServiceNow instance with REST Table API access.

API cheatsheet

Every loader exposes the same interface:

loader.load()                         # list[SnowDocument]
loader.lazy_load()                    # generator
loader.load_since(datetime_cutoff)    # list[SnowDocument]
loader.concurrent_load(max_workers)   # threaded
loader.concurrent_lazy_load(...)      # threaded generator

loader.load(verify=True)              # raises if the sweep lost records
loader.load(on_error="skip")          # finish past a dead page instead of aborting
loader.load(keyset=True)              # cursor paging, no offsets
loader.load(limit=5)                  # just enough to look at, one request
loader.load(keyset=True, checkpoint=FileCheckpoint("state.json"))   # resumable

On the connection, a long sweep can be made resumable:

conn.get_records("cmdb_ci", keyset=True, checkpoint=FileCheckpoint("sweep.json"))
conn.concurrent_get_records("incident", max_workers=16, checkpoint=FileCheckpoint("inc.json"))

And the same work from a shell:

Credentials come from SNOW_INSTANCE, SNOW_USER and SNOW_PASS when the matching option is not given, so a password need not reach a shell history or a process list. Exit status is 0 when the sweep finished and verified, 1 when it did not return every record, 2 on a usage or credential problem, and 130 when interrupted.

Async siblings (when installed with [async]) follow the same shape: aload, alazy_load, aload_since. Every loader has one, and so does every LangChain and LlamaIndex adapter.

Document metadata carries both halves of every reference field, so a record can be joined to what it points at:

doc.metadata["assignment_group"]         # 'Service Desk'
doc.metadata["assignment_group_sys_id"]  # 'd625dcce...'
doc.metadata["priority"]                 # '5 - Planning'
doc.metadata["priority_value"]           # '5'

Pick the right pagination path

Three concurrency models, three jobs, plus a cursor mode that cuts across all of them. The numbers below came out of a run against a developer instance, not an estimate. Reproduce them on your own instance with scripts/benchmark_v030.py.

Two things worth taking from that chart. Throughput peaks at 16 workers, which is why 16 is the default, and 32 workers came out slower than 16 rather than marginally faster. Page size in the low hundreds measured best on this path; raising it does not buy speed here, because throughput is bounded by how many pages are in flight rather than by request count.

The async path pulls the other way and wants larger pages, which is why its default is 500. Do not carry a page size from one path to the other.

Separately from concurrency, keyset=True pages on a sys_id cursor rather than an offset. It is what makes a sequential run resumable, and it is immune to the tied-sort problem by construction rather than by convention. It is not a speed feature: on a 2,919 row table, an offset of 2,800 was no slower than an offset of 0, so the usual deep-offset argument did not reproduce at that scale. Measure your own table before choosing it for throughput.

The threaded path uses a per-thread requests.Session, which keeps connection pools and TLS state isolated per worker and avoids the connection-reuse failures some ServiceNow front ends exhibit when many concurrent requests share one session.

Code recipes

from snowloader import SnowConnection, SweepIncompleteError

with SnowConnection(
    instance_url="https://yourcompany.service-now.com",
    username="api_user",
    password="api_pass",
    page_size=100,          # smaller pages parallelise better
    order_by="sys_id",      # unique key, and the cheapest sort
) as conn:
    try:
        records = list(
            conn.concurrent_get_records(
                "cmdb_ci",
                max_workers=16,
                verify=True,        # free here: the count is already fetched
                on_error="skip",    # finish the run, then complain
            )
        )
    except SweepIncompleteError as exc:
        alert(f"CMDB extract incomplete: {exc.report}")
        # table=cmdb_ci expected=30000 returned=29900 distinct=29900
        # missing=100 duplicated=0 failed_pages=1
        raise

from snowloader import SnowConnection, TableLoader, RelationshipLoader

with SnowConnection(
    instance_url="https://yourcompany.service-now.com",
    username="api_user",
    password="api_pass",
    display_value="all",
    order_by="sys_id",
) as conn:
    for doc in TableLoader(conn, table="cmdb_ci").lazy_load(verify=True):
        graph.add_node(
            doc.metadata["sys_id"],
            name=doc.metadata.get("name"),
            # the value half, so this is the table name the CI lives in,
            # not the pretty label
            ci_class=doc.metadata.get("sys_class_name_value"),
        )

    for doc in RelationshipLoader(conn).lazy_load(verify=True):
        graph.add_edge(
            doc.metadata["parent_sys_id"],
            doc.metadata["child_sys_id"],
            type=doc.metadata["type"],
        )

from snowloader import SnowConnection, IncidentLoader

with SnowConnection(
    instance_url="https://yourcompany.service-now.com",
    username="api_user",
    password="api_pass",
) as conn:
    loader = IncidentLoader(connection=conn, query="active=true^priority<=2")
    for doc in loader.lazy_load():
        process(doc)

from snowloader import SnowConnection, IncidentLoader

with SnowConnection(
    instance_url="https://yourcompany.service-now.com",
    username="api_user",
    password="api_pass",
    page_size=100,
) as conn:
    total = conn.get_count("incident", query="state=6^close_notesISNOTEMPTY")

    for record in conn.concurrent_get_records(
        table="incident",
        query="state=6^close_notesISNOTEMPTY",
        max_workers=16,
    ):
        process(record)

    loader = IncidentLoader(connection=conn, query="state=6^close_notesISNOTEMPTY")
    docs = loader.concurrent_load(max_workers=16)

import asyncio
from snowloader import AsyncSnowConnection, AsyncIncidentLoader

async def main() -> None:
    async with AsyncSnowConnection(
        instance_url="https://yourcompany.service-now.com",
        username="api_user",
        password="api_pass",
        page_size=500,
        concurrency=16,
    ) as conn:
        loader = AsyncIncidentLoader(connection=conn, query="active=true")
        async for doc in loader.alazy_load():
            print(doc.page_content[:200])

asyncio.run(main())

from snowloader import SnowConnection
from snowloader.adapters.langchain import ServiceNowIncidentLoader

conn = SnowConnection(
    instance_url="https://yourcompany.service-now.com",
    username="api_user",
    password="api_pass",
)
loader = ServiceNowIncidentLoader(connection=conn, query="active=true")
docs = loader.load()  # list[langchain_core.documents.Document]

# Plug straight into any vector store
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

vectorstore = FAISS.from_documents(docs, OpenAIEmbeddings())

from snowloader.adapters.llamaindex import ServiceNowIncidentReader

reader = ServiceNowIncidentReader(connection=conn, query="active=true")
docs = reader.load_data()  # list[llama_index.core.schema.Document]

from llama_index.core import VectorStoreIndex
index = VectorStoreIndex.from_documents(docs)

from datetime import datetime, timezone

loader = IncidentLoader(connection=conn)
docs = loader.load()                          # First run: everything
last_sync = datetime.now(timezone.utc)

updated = loader.load_since(last_sync)        # Subsequent runs: only changes

from snowloader import CMDBLoader

loader = CMDBLoader(
    connection=conn,
    ci_class="cmdb_ci_server",
    include_relationships=True,
)

for doc in loader.lazy_load():
    # -> db-prod-01 (Depends on::Used by)
    # <- load-balancer-01 (Depends on::Used by)
    print(doc.page_content)

loader = IncidentLoader(connection=conn, query="active=true", include_journals=True)
for doc in loader.lazy_load():
    print(doc.page_content)
    # Incident: INC0000007
    # Summary: Need access to sales DB
    # ...
    # [work_notes] 2024-06-01 09:15:00 by alice
    # Restarted Exchange service, monitoring.

from snowloader import AttachmentLoader

# Metadata only
loader = AttachmentLoader(connection=conn, query="table_name=kb_knowledge")
for doc in loader.lazy_load():
    print(doc.metadata["file_name"], doc.metadata["size_bytes"])

# Download a specific file by sys_id
loader.download_to("att_sys_id", "./out/diagram.png")

# Eager download with size cap (10 MB)
loader = AttachmentLoader(connection=conn, download=True, max_size_bytes=10 * 1024 * 1024)
for doc in loader.lazy_load():
    blob = doc.metadata.get("content_bytes")

# Basic Auth (development)
conn = SnowConnection(instance_url="...", username="admin", password="pass")

# OAuth Client Credentials (recommended for production)
conn = SnowConnection(instance_url="...", client_id="...", client_secret="...")

# OAuth Password Grant
conn = SnowConnection(instance_url="...", client_id="...", client_secret="...",
                       username="...", password="...")

# Bearer Token (pre-obtained)
conn = SnowConnection(instance_url="...", token="eyJhbG...")

snowloader extract incident --out incidents_closed.jsonl --resume \
    --query "stateIN6,7^close_notesISNOTEMPTY^sys_updated_on>=javascript:gs.daysAgoStart(730)" \
    --fields sys_id,number,short_description,close_notes,state,priority,category,opened_at,resolved_at \
    --display-value all --workers 16

snowloader extract incident --out incidents_active.jsonl --resume \
    --query "active=true" --display-value all --workers 16

import json
from snowloader import FileCheckpoint, SnowConnection, SweepIncompleteError

QUERY = "stateIN6,7^close_notesISNOTEMPTY"
checkpoint = FileCheckpoint("incidents_closed.state.json")

with SnowConnection(
    instance_url="https://yourcompany.service-now.com",
    username="api_user",
    password="api_pass",
    page_size=100,
    display_value="all",
) as conn:
    try:
        with open("incidents_closed.jsonl", "a", encoding="utf-8") as fh:
            for record in conn.concurrent_get_records(
                "incident", query=QUERY, max_workers=16,
                checkpoint=checkpoint, verify=True,
            ):
                fh.write(json.dumps(record, ensure_ascii=False) + "\n")
    except SweepIncompleteError as exc:
        alert(f"incident extract incomplete: {exc.report}")
        raise

Configuration

AsyncSnowConnection takes the same arguments plus concurrency (default 16) and keep_alive (default False, which trades a TLS handshake per request for immunity to the empty-body responses some ServiceNow front ends return on reused connections under load). Measure both against your own instance before turning it on.

See the full documentation for every parameter.

Roadmap

Version	Feature	Status
v0.1	Six sync loaders, LangChain + LlamaIndex adapters, 4 auth modes, delta sync, journal entries, HTML cleaning, CMDB graph traversal
v0.2	Async support (aiohttp) and async variants of every loader and adapter
v0.2	Attachment loader for sys_attachment with optional eager download and size cap
v0.2	Threaded sync paginator (concurrent_get_records, concurrent_load) with per-thread sessions
v0.2	parse_labelled_int helper for fields like priority, urgency, impact
v0.3	Deterministic pagination: every ORDERBY chain ends in sys_id, so offset paging cannot lose rows to a tied sort column
v0.3	Sweep verification (verify=True, SweepReport, SweepIncompleteError) and a partial failure policy (on_error="skip")
v0.3	Reference fields as both halves everywhere, plus public field helpers (reference, raw_value, expand_reference_keys)
v0.3	Generic TableLoader, RelationshipLoader, aconcurrent_get_records, configurable order_by and since_field
v0.4	Keyset pagination (keyset=True) and resumable extractions (Checkpoint, FileCheckpoint)
v0.5	Command line: snowloader extract and snowloader count, with the careful choices as defaults
v0.6	limit on every loader, checkpoint on the loaders and the async path, and async adapters for the last two loaders
-	Direct vector store streaming. Removed: the LangChain and LlamaIndex adapters already reach dozens of stores, maintained by those projects
v1.0	Custom field mapping for heavily customized instances

Write support stays out of scope. A read-only guarantee is the reason somebody points this at production without raising a change, and a read library that starts writing has to grow opinions about Data Policies, choice lists and business rules or it becomes a polite way to corrupt a CMDB. If it ever happens it will be a sibling package sharing connection, auth and retry.

Contributing

Contributions are welcome.

1. Fork the repository
2. Create a feature branch
3. Write tests first (the project uses pytest + responses for HTTP mocking)
4. Ensure the quality gate passes:

   ruff check src/ tests/ && ruff format --check src/ tests/ && mypy src/snowloader/ && pytest tests/ -x
   

5. Open a pull request

Author

License

MIT. See LICENSE for the full text.