Implement a Custom Connector

A custom connector enables the C3 AI Platform to communicate with an external system that is not supported by the platform's built-in SQL, file, cloud-message, or API source types. A connector can support live query operations and, when combined with additional framework components, participate in Data Integration workflows. For more information about the framework's architecture and components, see Custom Connector Framework.

This topic describes how to create a custom connector for live query operations by defining the connector, defining a credential type, implementing the required methods, registering the connector, attaching credentials to the registered instance, and binding an application type to it.

Copy link to this section

Create a VirtualSourceSystem

Create a subtype of VirtualSourceSystem to implement the custom connector. This subtype contains the logic required to communicate with the external system. VirtualSourceSystem mixes SourceSystem, so the connector can also participate in Data Integration. This topic covers the live query hooks. Write hooks belong in the write-back topic.

This pattern applies to non-REST connectors. For REST-based external systems, mix VirtualRestSourceSystem and use VirtualRestSourceSystem.Config instead of VirtualSourceSystem.Config directly. See Implement a REST-Based Connector.

VirtualSourceSystem takes six type parameters:

• C: the configuration type.
• CREDS: the credential type.
• REQ: the read request type.
• BATCH: one unit of read work.
• WSPEC: the write request type. The write-back topic covers it.
• R: the row representation, Obj or Arrow.

REQ mixes VirtualSourceSystem.Read.Spec and the capability mixins the connector supports. The platform copies onto the request only the fields those mixins declare:

• Read.Filterable declares filter.
• Read.Limitable declares limit.
• Read.Offsetable declares offset.
• Read.Orderable declares order.
• Read.Includable declares include.
• Read.Projectable declares projection.
• Read.Joinable declares joins.
• Read.Groupable declares group and having.
• Read.Arrowable declares batchSize.

A connector that reads each request as one unit uses the same type for REQ and BATCH. That type mixes VirtualSourceSystem.Read.BatchSpec. A connector that splits a request binds its own batch type and returns those batches from batches.

readStream returns rows as R. Bind R to Obj or Arrow. The platform bridges Obj and Arrow when the caller needs the other representation. A custom R must implement rToObj or rToArrow so a read can be bridged. Saving the connector also requires objToR or arrowToR, and objsToR or arrowsToR. The write-back topic covers those write hooks.

Reads go through two hooks on VirtualSourceSystem.BatchReader:

• readStream(type, batch) returns stream<R>. It backs fetch, fetchObjStream, and fetchArrowStream.
• readTupleStream(type, batch) returns stream<Tuple>. It backs evaluate. It is the only read hook that accepts a null type, which the platform uses for schema inference.

The connector must also implement batches, singleBatch, or both. These hooks describe how a request becomes batches. Saving the connector fails when neither hook is implemented. Validating its metadata fails for the same reason.

fetchCount and exists call countObjs. They do not count rows from readStream unless your countObjs implementation does that. Those calls fail when countObjs is not implemented.

Only ping is required for every connector. Also implement singleBatch or batches. Implement readStream, readTupleStream, and countObjs for the live query operations the connector supports.

type MyConnector mixes VirtualSourceSystem<
    MyConnector.Config,
    MyConnector.Credentials,
    MyConnector.ReadSpec,
    MyConnector.ReadSpec,
    VirtualSourceSystem.Write.Spec,
    Obj> {

    readStream:      ~ js-server // optional; required to support fetch()
    readTupleStream: ~ js-server // optional; required to support evaluate() and schema inference
    countObjs:       ~ js-server // optional; required to support fetchCount() and exists()
    singleBatch:     ~ js-server // required, or batches: read the request as one unit
    ping:            ~ js-server // required
}

type MyConnector.ReadSpec mixes VirtualSourceSystem.Read.BatchSpec,
    VirtualSourceSystem.Read.Filterable,
    VirtualSourceSystem.Read.Limitable,
    VirtualSourceSystem.Read.Offsetable,
    VirtualSourceSystem.Read.Orderable,
    VirtualSourceSystem.Read.Includable

To support evaluate, also mix EvaluateSpec into MyConnector.ReadSpec. The platform then copies projection, group, having, and joins onto the request. Platform samples such as SampleReadSpec mix FetchSpec and EvaluateSpec when the connector honors those specs. The capability mixins above declare only the fields you support.

The connector specifies:

• A configuration type.
• A credential type used to authenticate with the external system.
• The request type, the batch type, and the row representation.
• The methods that execute queries and verify connectivity.

If the connector requires configuration beyond credentials, define the additional configuration properties in the connector's configuration type.

Copy link to this section

Bind Application Types with VirtualPersistable

Mix VirtualPersistable into an application type to route its persistence operations to the connector. VirtualPersistable mixes External, Fetchable, Evaluatable, and Writable. The concrete type declares its own id field and does not use a C3-managed datastore.

The platform routes fetch and fetchObjStream to readStream. It routes evaluate to readTupleStream. It routes fetchCount and exists to countObjs. Create, update, upsert, merge, and remove use the connector's write hooks. removeAll calls deleteAll. The write-back topic covers those hooks.

The platform resolves which connector instance to use from the type's @connector(sourceSystem=...) annotation. For the concrete binding syntax, see Configure Connector Binding, later in this topic.

Copy link to this section

Create Credentials

Define a credential type by mixing VirtualSourceSystem.Credentials. See Custom Connector Credentials for more information.

type MyConnector.Credentials mixes VirtualSourceSystem.Credentials {
    name: string
    endpoint: string
    @config(secret=true)
    authToken: string

    forName: ~ js-server
}

type MyConnector.Config mixes
    VirtualSourceSystem.Config<MyConnector.Credentials>

Include the properties required to authenticate with the external system. Mark sensitive values with @config(secret=true) so they are stored and managed securely by the platform. Because VirtualSourceSystem.Credentials mixes Named, a concrete credential type must also implement forName and provide a name value, which is the key used to store and look up the credential set in the connector's virtualCredentials map.

The platform masks secret values for a caller who cannot read secrets. Rewriting stored secrets requires an environment admin or a cluster admin. See Custom Connector Credentials for details on credential storage, retrieval, and permissions.

The connector instance uses the configured credentials when communicating with the external system. Credentials are attached to the connector instance after registration; see Attach Credentials to the Instance, later in this topic.

Copy link to this section

Implement Connector Methods

Implement the methods required for the connector to communicate with the external system.

Copy link to this section

Implement singleBatch

Implement singleBatch when the connector reads the whole request as one unit. Return the request. Implement batches instead when the connector splits the request into smaller units. Implement at least one of the two hooks.

function singleBatch(type, req) {
    return req;
}

Copy link to this section

Implement readStream

Implement readStream to read one batch as the connector's element type. Fetch operations against a VirtualPersistable type, such as MyWidget.fetch(...), are routed to this method.

The method receives the requested type and one batch. filter, order, limit, offset, and include are present only when MyConnector.ReadSpec mixes the matching capability types. They do not arrive automatically from a FetchStreamSpec. The platform still cuts the returned rows to the caller's limit when the request does not declare limit. Translate the fields the request declares into the external system's query language, and return the matching objects as a stream.

function readStream(type, batch) {
    // Translate batch.filter, batch.order, batch.limit, and batch.offset
    // into the external system's native query language.
    var records = queryExternalSystem(type, batch.filter, batch.order, batch.limit, batch.offset);
    var index = 0;
    return C3.Stream.make(type.meta().referenceType(), function () {
        if (index >= records.length) {
            return null;
        }
        return type.make(records[index++]);
    });
}

Return null from the next function when the batch has no more rows. Build each object inside that function so the stream stays lazy.

Copy link to this section

Implement readTupleStream

Implement readTupleStream to read one batch as tuples. Evaluate operations against a VirtualPersistable type, such as MyWidget.evaluate(...), are routed to this method. Schema inference also calls this method, with type set to null.

function readTupleStream(type, batch) {
    // results, for example: [{ paymentType: "card", count: 42 }, { paymentType: "cash", count: 7 }]
    var results = queryExternalSystem(type, batch);
    var index = 0;
    return C3.Stream.make(Tuple.meta().referenceType(), function () {
        if (index >= results.length) {
            return null;
        }
        var r = results[index++];
        return Tuple.of("paymentType", "string", r.paymentType, "count", "int", r.count);
    });
}

Translate the batch into the external system's query language and return tuples to the platform. Build each tuple with Tuple.of (name, value type, and value). The platform converts those tuples into the CellTuple results the caller receives.

When type is null, read the external target from batch.table. For a typed read, type.table() returns the external table. That value is @connector(table) when set, and the C3 type name otherwise.

The connector computes the evaluation itself. Mix EvaluateSpec into the request type so the batch can carry projection, group, having, and joins.

Copy link to this section

Implement countObjs

Implement countObjs to support fetchCount and exists. The method receives the type and the read request, and returns the number of matching records.

Use a native count query when the external system provides one. A connector that can only stream rows counts the readStream result in countObjs. The platform does not perform that count.

function countObjs(type, req) {
    return readStream(type, req).size;
}

Copy link to this section

Implement ping

Implement ping to verify that the connector can successfully communicate with the external system.

function ping() {
    var creds = this.getCredentialsByKey("default", true);
    try {
        checkConnection(creds.endpoint);
        return PingResult.make({
            url: creds.endpoint,
            reachable: true
        });
    } catch (e) {
        return PingResult.make({
            url: creds.endpoint,
            reachable: false,
            errorMsg: e.message
        });
    }
}

PingResult.url is required. Set it to the endpoint that was checked, and set errorMsg when reachable is false. The platform uses this operation to validate connectivity between the connector and the external system.

Copy link to this section

Register the Connector

After implementing the connector, create and register an instance of MyConnector.

MyConnector.make({
    name: "MyConnector"
}).upsert();

The registered connector instance represents the configured connection to the external system and is referenced when application types perform live query operations.

{
  "type": "MyConnector",
  "name": "MyConnector"
}

During package provisioning, the platform automatically creates the connector instance. Connector instances provisioned this way are managed by the application package; check with your platform administrator on the permissions and process for updating credentials or configuration on a package-provisioned instance at runtime.

Copy link to this section

Attach Credentials to the Instance

Registering the connector instance does not attach credentials to it. Before ping or any live query operation can succeed, set credentials on the registered instance with setCredentials:

MyConnector.forName("MyConnector").setCredentials(
    MyConnector.Credentials.make({
        name: "default",
        endpoint: "https://api.example.com",
        authToken: "..."
    }),
    false
);

The merge argument (false above) controls whether the call replaces all credentials on the instance or merges the provided fields into any existing credential set with the same name. See Custom Connector Credentials for more information on storing, retrieving, and merging credentials.

Copy link to this section

Configure Connector Binding

Bind an application type to the registered connector instance by annotating the type with @connector(sourceSystem=...) and mixing in VirtualPersistable. Declare id as string. VirtualPersistable does not generate it.

@connector(sourceSystem="MyConnector")
type VirtualWidget mixes VirtualPersistable {
    id: string
    name: string
    category: string
    price: double
}

The sourceSystem value must match the registered connector instance's name, the value passed to make when the instance was created (see Register the Connector). It does not have to match the connector type's name. This example registers the instance as MyConnector, so the type binds to @connector(sourceSystem="MyConnector"). The two names can differ. Registering MyConnector.make({name: "myConnectorInstance"}) requires @connector(sourceSystem="myConnectorInstance").

@connector also accepts table on the type and fieldName on a field. Set table when the external table, collection, or REST path differs from the C3 type name. The platform copies table onto the read request. When table is unset, the platform uses the C3 type name. Set fieldName when the external column name differs from the C3 field name. The connector reads fieldName from the annotation. When fieldName is unset, use the C3 field name. The platform does not translate field names.

@connector(sourceSystem="MyConnector", table="WIDGETS")
type VirtualWidgetMapped mixes VirtualPersistable {
    id: string

    @connector(fieldName="widget_name")
    name: string

    @connector(fieldName="widget_category")
    category: string

    @connector(fieldName="unit_price")
    price: double
}

When an application performs a live query against the bound type, the platform resolves the configured connector instance and invokes the connector implementation to retrieve data from the external system.

Copy link to this section

Full Example (Python)

The following example implements the same connector shown throughout this topic using Python (py-server) instead of JavaScript.

type MyPythonConnector mixes VirtualSourceSystem<
    MyPythonConnector.Config,
    MyPythonConnector.Credentials,
    MyPythonConnector.ReadSpec,
    MyPythonConnector.ReadSpec,
    VirtualSourceSystem.Write.Spec,
    Obj> {

    readStream:      ~ py-server
    readTupleStream: ~ py-server
    countObjs:       ~ py-server
    singleBatch:     ~ py-server
    ping:            ~ py-server
}

type MyPythonConnector.ReadSpec mixes VirtualSourceSystem.Read.BatchSpec,
    VirtualSourceSystem.Read.Filterable,
    VirtualSourceSystem.Read.Limitable,
    VirtualSourceSystem.Read.Offsetable,
    VirtualSourceSystem.Read.Orderable,
    VirtualSourceSystem.Read.Includable

type MyPythonConnector.Credentials mixes VirtualSourceSystem.Credentials {
    name: string
    endpoint: string
    @config(secret=true)
    authToken: string

    forName: ~ py-server
}

type MyPythonConnector.Config mixes
    VirtualSourceSystem.Config<MyPythonConnector.Credentials>

def readStream(this, type, batch):
    # Translate batch.filter, batch.order, batch.limit, and batch.offset
    # into the external system's native query language.
    records = queryExternalSystem(type, batch.filter, batch.order, batch.limit, batch.offset)

    def gen():
        for r in records:
            yield type.make(r)

    result = gen()
    return c3.Stream.make(type.meta().referenceType(), lambda: next(result, None))


def readTupleStream(this, type, batch):
    # results, for example: [{"paymentType": "card", "count": 42}, {"paymentType": "cash", "count": 7}]
    results = queryExternalSystem(type, batch)

    def gen():
        for r in results:
            yield c3.Tuple.of("paymentType", "string", r["paymentType"], "count", "int", r["count"])

    result = gen()
    return c3.Stream.make(c3.Tuple.meta().referenceType(), lambda: next(result, None))


def countObjs(this, type, req):
    return readStream(this, type, req).size


def singleBatch(this, type, req):
    return req


def ping(this):
    creds = this.getCredentialsByKey("default", True)
    try:
        checkConnection(creds.endpoint)
        return c3.PingResult.make({"url": creds.endpoint, "reachable": True})
    except Exception as e:
        return c3.PingResult.make({"url": creds.endpoint, "reachable": False, "errorMsg": str(e)})


def forName(cls, name):
    return cls.make({"name": name})

Register, attach credentials to, and bind the connector the same way as the JavaScript example:

c3.MyPythonConnector.make({"name": "MyPythonConnector"}).upsert()

c3.MyPythonConnector.forName("MyPythonConnector").setCredentials(
    c3.MyPythonConnector.Credentials.make({
        "name": "default",
        "endpoint": "https://api.example.com",
        "authToken": "..."
    }),
    False
)

@connector(sourceSystem="MyPythonConnector")
type PythonWidget mixes VirtualPersistable {
    id: string
    name: string
    category: string
    price: double
}