Notebook Reference

Everything a Custom connector notebook can use. For how notebooks work — running cells, previewing, packages and secrets — start with Custom Connectors.

Two names are available in every cell, with or without the import:

from classifyre import Asset, ctx

The functions you define

test_connection() -> dict

Called by Test connection, and once at the start of every scan — a scan stops before ingesting anything if this fails.

def test_connection() -> dict:
    response = httpx.get(f"{ctx.var('api_base')}/health", timeout=10)
    if response.is_success:
        return {"status": "SUCCESS", "message": "API reachable."}
    return {"status": "FAILURE", "message": f"HTTP {response.status_code}"}

Return {"status": "SUCCESS" | "FAILURE", "message": str}. Anything else is treated as a failure, with a message saying so. Raising works too — the traceback becomes the failure message.

extract()

The heart of a connector. Yields the things to scan.

def extract():
    for row in api.list(offset=ctx.offset, limit=ctx.limit):
        yield Asset(id=str(row["id"]), name=row["title"], content=row["body"])

Yield as you go. A generator lets a large source start being scanned before it has finished listing; building a list first means holding all of it in memory and waiting.

discover() -> dict (optional)

A summary of what this source can see — indexes, projects, buckets. Free-form.

fetch_content(asset_id) -> tuple[str, str] | None (optional)

Content on demand for one asset. Only reached when the asset’s content wasn’t already supplied by extract(), so most connectors never need it. Return (raw, text).

Asset

What extract() yields.

Linking assets

links takes the ids you already use, and Classifyre resolves them into the investigation graph:

yield Asset(
    id="ticket-42",
    name="Login fails on mobile",
    content=body,
    links=["ticket-17", "user-8801"],   # ids, not hashes
)

Lineage

links says two assets are connected. It does not say how, and that turns out to be the whole question: an attachment, a foreign key and a derived table are three different relationships, and only one of them answers “if I change this, what breaks?”.

Define a relationships() function and pick the builder that matches what you mean. Assets are named by the same ids you gave them in extract().

from classifyre import Ref, FieldMapping, FlowType, flow, contains, references, same_as
 
def relationships():
    # Lineage. The only class an impact question follows.
    yield flow(
        upstream=Ref.asset("orders"),
        downstream=Ref.asset("top_deliveries"),
        type=FlowType.TRANSFORM,
        fields=[
            FieldMapping("delivery_time", ["placed_on", "delivered_on"],
                         "DATEDIFF(placed_on, delivered_on)"),
            # A null downstream is an *indirect* dependency: it shaped which
            # rows came out without feeding any one output column.
            FieldMapping(None, ["placed_on"]),
        ],
    )
 
    yield contains(Ref.asset("orders"), Ref.asset("orders_2024"))   # part of
    yield references(Ref.asset("orders"), Ref.asset("customers"))   # points at

flow takes both ends as keyword arguments on purpose. A reversed lineage edge is silently wrong rather than loudly broken, so it is not something you can write by accident.

Lineage across systems

Ref.urn(...) names an object by what its platform calls it, so you can point at a table this connector does not produce and has never seen. If a source for that system is scanned — before this run or a month after it — the two halves find each other and the edge completes itself.

from classifyre import urn_for
 
yield flow(
    upstream=Ref.urn(urn_for("snowflake", "acme", "PROD", "PUBLIC", "RAW_ORDERS")),
    downstream=Ref.asset("orders"),
    type=FlowType.COPY,
)

Build the URN with urn_for(...) rather than writing the string yourself: each connector folds capitalisation and default ports its own way, and a URN that differs by one letter never matches. Setting Asset.urn does the same thing in the other direction — it lets another source’s lineage point at your asset.

Start from the Lineage and links template, which has all of this running.

Files and images

Set content_bytes and a connector that fetches files gets everything a built-in file source gets: text extraction, normalized file metadata, and the binary and image detectors.

def extract():
    for item in bucket.list():
        yield Asset(
            id=item.key,
            name=item.filename,
            content_bytes=bucket.download(item.key),
            mime_type=item.content_type,   # detected if you omit it
            kind="file",
        )

You don’t need to parse anything — a PDF’s text, an image’s dimensions and the file’s size all get filled in for you.

parse and pages

from classifyre import pages, parse — the same extractor every built-in file source uses, so a notebook never implements format handling.

parse(source, *, name="", mime_type=None) -> ParsedContent

source may be a path, bytes, an open binary handle, or a NotebookFile from ctx.files. mime_type is a hint, trusted over sniffing when you already know it — useful for a payload an API declared.

It never raises. A corrupt or unreadable file comes back with .error set and empty text, so one bad attachment costs one asset rather than the run. The object is falsey when .error is set:

parsed = parse(path)
if not parsed:
    ctx.log(f"skipping {path.name}: {parsed.error}")
    continue

Handled formats include PDF, DOCX, XLSX, PPTX, the OpenDocument formats, EML and MSG, RTF, HTML, JSON, CSV/TSV, Parquet and Arrow, archives, and images — images and scanned PDFs are OCR’d.

pages(source, *, page_size=100) -> Iterator[str]

Reads a payload a page at a time instead of whole: rows for a tabular file, lines for everything else. A multi-gigabyte dump becomes many assets and is never in memory at once.

for index, page in enumerate(pages(ctx.file("dump.parquet"), page_size=500)):
    yield Asset(id=f"dump-{index}", content=page, kind="table")

Files the source carries

ctx.files — uploaded files

Files uploaded to the source, downloaded to local disk before any cell runs. Empty is a normal state, not an error. Available in every deployment.

Each entry is a NotebookFile:

ctx.folders — local folders (desktop only)

Folders configured on the source, as pathlib.Path. Nothing is copied — the notebook opens files where they are.

Empty in a Kubernetes deployment: there is no such machine there, and the API refuses to save a source that carries folders. Upload the files instead.

ctx

Configuration

This run

Resuming with your own cursor

If your system pages by something other than a count — a token, a timestamp — record it yourself and it wins over the positional offset:

def extract():
    token = ctx.cursor.get("page_token")
    page = api.list(page_token=token, limit=ctx.limit)
 
    for row in page.items:
        yield Asset(id=str(row["id"]), name=row["title"], content=row["body"])
 
    ctx.set_cursor({"page_token": page.next_token})

Long-running work

ctx.should_abort turns True when someone presses Stop or the run is cancelled. Python can’t be interrupted from outside, so a loop that never checks it only ends when the execution times out.

for row in huge_result_set:
    if ctx.should_abort:
        return
    yield Asset(...)

Limits

Set under the notebook’s options:

Rules worth knowing

• Cells must be valid standard Python. IPython magics (%time, !pip install) are rejected, because the notebook has to run as an ordinary module in production.
• Top-level functions only. test_connection and extract have to be defined at the top level of a cell — a function nested inside a class or another function isn’t found.
• Packages are declared, not installed in code. Use the packages table rather than a pip install cell; the declared list is what gets installed before your cells run.

When something goes wrong

The schema behind all of this — every field, with types and examples — is at Custom Connector.