Data Sources

A data source lets a plugin pull documents from an external service into a Noxus knowledge base. You implement one method — fetch — that gathers the documents and returns them as files; the platform ingests them.

The class

Subclass BaseDataSource[Config] and implement fetch:

from noxus_sdk.datasources import BaseDataSource, DatasourceConfiguration
from noxus_sdk.ncl import Parameter, ConfigText
from noxus_sdk.files import File
from noxus_sdk.plugins.context import RemoteExecutionContext


class DriveFolderConfig(DatasourceConfiguration):
    folder_id: str = Parameter(
        default="",
        description="Folder to ingest",
        display=ConfigText(label="Folder ID"),
    )


class DriveFolder(BaseDataSource[DriveFolderConfig]):
    datasource_name = "DriveFolder"      # stable id, unique within the plugin
    title = "Drive folder"
    description = "Ingest every document in a Drive folder"
    integrations = ["gdrive"]            # credentials this source needs

    async def fetch(self, ctx: RemoteExecutionContext) -> list[File]:
        client = _client_from_ctx(ctx)
        helper = ctx.get_file_helper()

        files: list[File] = []
        for doc in await client.list_folder(self.config.folder_id):
            content = await client.download(doc["id"])          # bytes
            file = await helper.upload_file(
                content, name=doc["name"], content_type=doc["mime"]
            )
            files.append(file)
        return files

def datasources(self) -> list[type[BaseDataSource]]:
    return [DriveFolder]

How `fetch` works

Gather the documents

Read whatever you need from the external service, using self.config and the integration credentials.

Upload each file's bytes

Persist content with ctx.get_file_helper().upload_file(bytes, name=..., content_type=...). The upload goes over the host callback — the bytes are not returned inline — and you get back a File descriptor.

Return the files

Return the list[File]. The platform ingests them into the knowledge base.

Ingestion is one-shot today: fetch returns the current set of documents each time it runs. Incremental sync (only new/changed documents) is a later phase — supports_sync defaults to False.

Reading credentials

List the integration types in integrations and read them from the context:

creds = ctx.get_integration_credentials("gdrive") or {}
token = creds.get("access_token", "")

See Creating integrations for the credential model, and Working with files for the File helpers.

Definition fields

Field	Required	Notes
`datasource_name`	yes	Stable id, unique within the plugin.
`title` / `description`	no	Shown in the KB source picker; `title` defaults to `datasource_name`.
`integrations`	no	Credential types the source reads (`[]` if none).
`supports_sync`	no	Incremental sync — `False` for now (one-shot fetch).
`image`	no	Icon URL.

Overview

Tutorial

Reference

GitOps

The class

How `fetch` works

Reading credentials

Definition fields

​The class

​How fetch works

​Reading credentials

​Definition fields

The class

How `fetch` works

Reading credentials

Definition fields