Notebook Reference

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

Every name lives in one module, and is pre-bound in every cell whether or not you import it:

from classifyre import (
    Asset, ctx,                                       # the two you always use
    parse, pages,                                     # reading files
    Ref, FieldMapping, FlowType, urn_for,             # naming things
    flow, contains, references, same_as, uses,        # relationships
)

The shortest connector that works

Two functions. Everything else on this page is optional.

from classifyre import Asset, ctx
 
 
def test_connection() -> dict:
    return {"status": "SUCCESS", "message": "Ready."}
 
 
def extract():
    for row in api.list():
        yield Asset(id=str(row["id"]), name=row["title"], content=row["body"])

The functions you define

They must be defined at the top level of a code cell. A function nested inside a class or another function is not found.

test_connection() -> dict

A scan stops before ingesting anything if this fails, so make it actually reach the system.

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):
        if ctx.should_abort:
            return
        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. Yielding a dict of Asset fields works too, but you lose the validation Asset does.

discover() -> dict (optional)

A summary of what this source can see — indexes, projects, buckets. Free-form; a non-dict return is wrapped as {"result": ...}.

def discover() -> dict:
    return {
        "projects": [p.key for p in api.projects()],
        "total_issues": api.count(),
    }

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

Content on demand for one asset. Only reached when the asset’s content was not already supplied by extract(), so most connectors never need it. Return (raw, text), or one string used as both.

def fetch_content(asset_id: str):
    body = api.get_body(asset_id)
    if body is None:
        return None
    return (body.html, body.plain_text)

Reach for it when listing is cheap and fetching is not: extract() yields metadata for ten thousand documents, and only the ones that survive sampling get downloaded.

relationships() (optional)

Typed edges between assets. See Relating assets.

def relationships():
    yield flow(upstream=Ref.asset("orders"), downstream=Ref.asset("daily_totals"))

Yield only edges built by flow, contains, references, same_as or uses. A raw dict is rejected on purpose — the value of this function is that you had to say which relationship you mean.

Asset

What extract() yields. One thing worth scanning.

Choosing kind

kind decides how the asset is presented and which metadata fields apply. An unrecognised value is corrected to record with a warning rather than failing the scan — losing a four-hour run at asset 40,000 over a typo is the worse outcome.

Choosing content_type

Leave it unset unless the payload is not plain text. It routes which detectors run — set it wrong and image detectors run over text, or never run at all.

TXT · TABLE · IMAGE · VIDEO · AUDIO · URL · BINARY · OTHER

If you set content_bytes, the type is detected from the bytes and you can skip this entirely.

Files and images

Set content_bytes and the connector 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 do not need to parse anything here — a PDF’s text, an image’s dimensions and the file’s size are all filled in for you. Parse it yourself only when you want the text before yielding, for example to skip empty documents.

Reading files: parse and pages

The same extractor every built-in file source uses, so a notebook never implements format handling.

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.

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

source may be a Path, a str 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
 
yield Asset(id=path.name, content=parsed.text, kind="file")

A PDF fetched over HTTP — bytes, with the declared type as a hint:

response = httpx.get(url)
parsed = parse(response.content, name="policy.pdf", mime_type="application/pdf")
yield Asset(id=url, name="policy.pdf", content=parsed.text, kind="document")

An .eml on disk — the parser reads headers, body and attachments:

for path in ctx.folder("mail").rglob("*.eml"):
    parsed = parse(path)
    yield Asset(id=str(path), name=path.name, content=parsed.text, kind="document")

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")

Use parse when the file fits comfortably in memory and you want one asset out of it. Use pages when it does not, or when each page is worth investigating on its own.

Relating assets

Two mechanisms, and the difference matters.

Asset.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?”

relationships() types the relationship. Reach for it whenever the kind of connection is part of the answer.

links — connected, unspecified how

Takes the ids you already use. Cheap, and enough when “these belong together” is all you mean.

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

Order does not matter — a link to an asset yielded later resolves fine. A link to an id this run never yields simply has nothing on the other end. Links are directional as written; link both ways when both directions are worth following.

The five builders

from classifyre import (
    Ref, FieldMapping, FlowType, flow, contains, references, same_as, uses,
)
 
 
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"]),
        ],
    )
 
    # A partition is part of its table — collapse them in the graph.
    yield contains(Ref.asset("orders"), Ref.asset("orders_2024"))
 
    # A foreign key points somewhere; it does not move values.
    yield references(Ref.asset("orders"), Ref.asset("customers"))
 
    # The export and the table are the same real thing under two names.
    yield same_as(Ref.asset("orders"), Ref.asset("orders_export"))
 
    # A person read it. Weight on the node, not a path through it.
    yield uses(Ref.asset("analyst-88"), Ref.asset("orders"))

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. FlowType is TRANSFORM (default), VIEW, COPY, WRITE, EXPORT or SEND.

Naming an endpoint: Ref

Declaring meaning: means

A connector often knows what a record is — the register says the legal form, the CRM says the account type. Say it with means() instead of leaving it for a binding to rediscover:

from classifyre import Ref, means
 
 
def relationships():
    yield means(
        Ref.asset(company_id),
        Ref.term("gmbh"),
        evidence={"field": "Rechtsform", "value": "GES"},
    )

The asset gets a declared semantic link to the concept (see Glossary). The key may not exist yet: the declaration is kept as a pending reference and links as soon as someone defines the term. Declarations to draft terms link once the term is approved. Keys are lower-case letters, digits, ., _ and -; Ref.term refuses anything else.

Declaring an entity: Entity

A register or a customer master does not merely mention ACME Holding GmbH; its record is ACME. Say so on the asset:

from classifyre import Asset, Entity
 
 
yield Asset(
    id=fn,
    urn=urn_for("firmenbuch", fn),
    content=record_text,
    entity=Entity(
        type="ORGANIZATION",
        name="ACME Holding GmbH",
        identifiers={"vat": "ATU12345678", "fn": "123456a"},
        aliases=["ACME Holding"],
        attributes={"seat": "Wien"},
    ),
)

The entity is created in the glossary (or updated), anchored to this record, and from then on every document in every source that carries the name or one of the identifiers is a mention of it.

The asset needs a urn (or the entity a key): it is what makes the declaration the same entity on the next scan instead of a new one. A declared entity is the register’s word, so it is approved from the start. A later scan only adds — names, identifiers and attributes an operator edited stay as they are, and an entity merged into another is declared onto the survivor. Entities must be on in the workspace; declarations made while they are off are applied when they are turned on.

Lineage across systems

Ref.urn(...) lets you 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,
)

urn_for(platform, authority, *path) — authority is the account, host, workspace or bucket; the rest is the path within it. Build the URN with it 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.

Tagging what you already know

Some facts do not need detecting. The system you are reading from already knows that a table holds cardholder data, that a folder is under legal hold, that a dataset came from a regulated jurisdiction. Running a classifier over the content to re-derive that can only lose information.

Asset.tags records those facts directly. Each key is the key of a Tag detector; the value is what you are asserting.

yield Asset(
    id="prod.payments.transactions",
    name="transactions",
    content=sample_csv,
    kind="table",
    tags={"cardholder_data": "primary-account-numbers"},
)

Each entry becomes a finding on the asset, carrying that detector’s label and severity, and behaves like any other finding from there: it appears on the asset, filters by severity, and can be cited in a case.

Build the dict from what the source told you, and skip a key rather than asserting an empty value:

def extract():
    for table in catalog.tables():
        tags = {}
        if table.classification:
            tags["cardholder_data"] = table.classification
        if table.legal_hold:
            tags["legal_hold"] = "retained-pending-litigation"
 
        yield Asset(id=table.id, name=table.name, content=table.sample, tags=tags)

Before this works, create the Tag detector — Detectors → New → Tag — and give it the key you write here.

When to tag and when to detect

You do not select Tag detectors on a source. Every active one is available to every Custom connector automatically, so a new tag needs no source change.

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:

def extract():
    for file in ctx.files:
        parsed = file.parse()
        if not parsed:
            ctx.log(f"skipping {file.name}: {parsed.error}")
            continue
        yield Asset(id=file.name, name=file.name, content=parsed.text, kind="file")

ctx.folders — local folders

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

client = ApiClient(
    base_url=ctx.var("api_base"),
    token=ctx.secret("api_token"),
    verify=ctx.var("verify_tls", "true") == "true",
)

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})

New items first

A built-in source reads what is new since the last run before it carries on with older items, and stops going over old ground once it has covered everything. ctx.newest_first gives a connector the same behaviour in one call:

def extract():
    for doc in ctx.newest_first(
        api.list_documents(),
        order=lambda doc: (doc["updated_at"], doc["id"]),
    ):
        yield Asset(id=doc["id"], name=doc["title"], content=doc["body"])

Pass the whole listing, in any order. order returns (rank, id): rank is what makes an item newer — a timestamp, an increasing number — and should only change when the item does; id is unique. Each run gets ctx.page_size items (or limit=). Its position is kept in the cursor beside whatever you store there, so you can use both. Calling it declares the run’s coverage partial.

When the settings change

A built-in source starts its sweep over when its configuration changes or a custom detector it uses is edited, retrained or deleted. Your cursor is yours — it may hold state that cannot be rebuilt — so Classifyre does not clear it. It tells you instead:

def extract():
    cursor = {} if ctx.config_changed else ctx.cursor
    ...

The flag stays set until a run finishes, so a run that fails before acting on it does not lose it. ctx.newest_first restarts by itself, and so does the positional offset Classifyre keeps for a connector that just yields everything.

Long-running work

ctx.should_abort turns True when someone presses Stop or the run is cancelled. Python cannot 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:

Fixed limits: a notebook holds at most 200 cells of 100,000 characters each, and may declare at most 50 packages.

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 is not 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.
• There is no persistent kernel. Running cell N re-runs the current source of cells 1 through N. State from an earlier run does not survive.
• id must be stable. Everything — findings, history, the scan cache, tags — hangs off it. A run that renumbers its ids looks like a source that deleted everything and created it again.

When something goes wrong

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