Elements: documentatiehub

Databases en warehouses

Er zijn starters voor onder andere PostgreSQL, MySQL/MariaDB, Microsoft SQL Server, ClickHouse, BigQuery, Snowflake, Amazon RDS, Redshift, Oracle, MongoDB, Elasticsearch/OpenSearch, Redis, Databricks, Trino/Presto en DuckDB.

Zakelijke SaaS

Kant-en-klare integraties omvatten Salesforce, HubSpot, Stripe, Shopify, Dynamics 365 / Dataverse, ServiceNow, Jira, Confluence, GitHub, GitLab, Google Sheets, Excel / OneDrive, SharePoint en SFTP.

Streaming en observability

Start met Kafka / Confluent, Google Pub/Sub, Amazon Kinesis, Azure Event Hubs, Prometheus, Grafana Loki of OpenSearch-logzoekopdrachten.

API-, sync- en bestandspatronen

Voorbeelden omvatten API-key, Basic Auth, HMAC, cursorpaginering, rate-limitafhandeling, OAuth refresh, async jobs, incrementele cursors en CSV-, XLSX-, Parquet-, NDJSON- en veilige ZIP-inname.

Governed Dashview-patronen

Gebruik starters voor queryDataSources, multi-source joins, rowsRef, schemavalidatie, deduplicatie, PII-redactie, datasource-appends, Parquet-exports, RAG-lookups, goedkeuringstaken, notificaties, apps, widgets en triggers.

Alle Elementsoorten

De catalogus bevat starters voor data, catalog, stream_data, transformer, detection, action, app, llm_tool, widget, trigger en server.

Editor reference

Alle editorvelden, filemanageracties, validatieregels en codecontract.

Runtime & execution reference

Publish/versioning, runtime vars/secrets, test execute payloads en logvelden.

Runtime builtin contracts

Per helper: signature, payload format, return shape, voorbeeldcode en scope-notities.

Basisroute

1. Maak een Element dat rows teruggeeft met def run(params):.
2. Test het Element met een kleine params-payload en publiceer daarna een versie.
3. Maak een datasource met type element en kies het gepubliceerde Element.
4. Vul vaste input in via elementParams, bijvoorbeeld een default country of limit.
5. Refresh of test de datasource zodat kolommen en voorbeeldrows zichtbaar zijn.
6. Koppel een tabel-, chart- of KPI-widget aan die datasource en gebruik de returned columns zoals bij elke andere datasource.

Hoe filters terug naar het Element gaan

• Dashboardfilters en form-widget velden worden eerst runtime filterwaarden op de pagina.
• Een Element-datasource kan die waarden als params ontvangen wanneer Filters als params doorgeven aan staat. Deze instelling is zichtbaar voor livestream-Element-datasources en voor parquet-snapshot wanneer Snapshot bij elk verzoek is gekozen.
• Gebruik allowedRequestParams / Toegestane doorgestuurde filtersleutels om expliciet te bepalen welke filterkeys naar het Element mogen.
• Gebruik requestParamMap als de dashboardfilterkey anders heet dan de param die je Element verwacht.
• In het Element lees je ze gewoon uit params, bijvoorbeeld params.get("country").
• Tijdens parquet refresh lopen geneste datasource reads nog steeds via het data-worker-agent helperpad met de parquetrefresh tenant-system materialisatiecontext en normale tenantscope.

Hoe reads draaien

• Element-code roept queryDataSources, query_datasources, dataSources.query of dataSources.query_rows aan.
• De Element-subprocess stuurt een gestructureerde helperrequest naar de parent data-worker-agent. De parent draait de geneste dataquery met de huidige tenant, gebruiker, access context, SQL safety rules en forced filters.
• Geef expliciete sources mee. Een source kan een datasource-ID zijn of een volledige source spec zoals {"type":"datasource","dataSourceId":"orders-ds","alias":"orders"}.
• SQL-tabelnamen komen uit de source alias wanneer die is meegegeven, anders uit de DuckDB-veilige datasource-ID/tabelnaam die de worker exposeert.

Rows lezen

def run():
    sources = [{
        "alias": "orders",
        "type": "datasource",
        "dataSourceId": "orders-ds",
    }]
    result = queryDataSources("""
        SELECT customer, SUM(revenue) AS revenue
        FROM orders
        GROUP BY customer
        ORDER BY revenue DESC
        LIMIT 100
    """, sources=sources, maxRows=100)
    return result.rows

Duurzame datasource writes

Gebruik appendDataSourceRows of dataSources.appendRows wanneer een Element JSON rows moet appenden naar een uplink/parquet-backed datasource. De write wordt gecontroleerd tegen de effectieve request access context voordat de worker hem naar uplink stuurt.

def run():
    result = appendDataSourceRows(
        "element-events",
        [{"event": "refresh_complete", "count": 42}],
        merge=True,
        meta={"source": "element"},
    )
    return {"inserted": result["inserted"]}

Resultaat en limieten

• Normale read-results zijn dict-like en hebben rows, columns, rowCount, .to_pandas() en .to_dataframe().
• maxRows geldt nog steeds voor geneste reads, inclusief asTable-materialisatie.
• Gebruik rowsRef, inline of file source specs wanneer je worker-produced artifacts via de helper chaint.
• Helpercalls vergroten geen tenant-, user-, datasource- of forced-filter-scope.

return: standaard output

• Gebruik return rows voor normale Elements, Element-datasources en kleine tot middelgrote snapshots.
• De return value moet JSON-serialiseerbaar zijn. Voor Element-datasources is dat normaal een lijst met dicts of een tabelreferentie zoals {"table": "output"}.
• Return geen wrapper zoals {"rows": rows, "meta": ...} als de downstream datasource een pure rows-array verwacht.

yield: rows stuk voor stuk produceren

• Gebruik yield row wanneer je rows kunt produceren zonder eerst alles in een lijst te verzamelen.
• Een generator-resultaat wordt door de runner als streambare row-output behandeld en naar een tijdelijk row-artifact geschreven.
• Gebruik in dezelfde entrypoint niet tegelijk yield en return rows; kies één outputmodus.

Vermijd volledige materialisatie

def run(params):
    # Minder goed: alle rows eerst in memory verzamelen.
    rows = list(load_external_rows(params))
    return [normalize(row) for row in rows]

Produceer rows incrementeel

def run(params):
    # Beter voor grote bronnen: per row verwerken.
    for row in load_external_rows(params):
        yield normalize(row)

Voorbeeld A: run(params)

def run(params):
    limit = int(params.get("limit", 5))
    country = params.get("country", "NL")
    rows = []
    for i in range(limit):
        rows.append({
            "rank": i + 1,
            "country": country,
            "value": (i + 1) * 100
        })
    return rows

Voorbeeld B: parameters verwerken

def run(params):
    label = params.get("label", "demo")
    multiplier = int(params.get("multiplier", 2))
    values = params.get("values") or [1, 2, 3]
    return [{
        "label": label,
        "value": value,
        "score": value * multiplier
    } for value in values]