TypeScript SDK resources

Use the TypeScript SDK to define Tinybird resources as code with full type inference. You can define datasources, pipes, endpoints, materialized views, copy pipes, sink pipes, and connections in TypeScript and sync them to Tinybird.

Project layout

The SDK scaffolds the following files by default:

  • lib/tinybird.ts
  • tinybird.config.mjs

Configure credentials

Create a .env.local file in your project root with the Tinybird token and host:

TINYBIRD_TOKEN=p.your_token_here
TINYBIRD_URL=https://api.tinybird.co

Datasources

Define datasources with schemas and engines:

import { defineDatasource, engine, t, type InferRow } from "@tinybirdco/sdk";

export const pageViews = defineDatasource("page_views", {
  description: "Page view tracking data",
  schema: {
    timestamp: t.dateTime(),
    pathname: t.string(),
    session_id: t.string(),
    country: t.string().lowCardinality().nullable(),
  },
  engine: engine.mergeTree({
    sortingKey: ["pathname", "timestamp"],
  }),
});

export type PageViewsRow = InferRow<typeof pageViews>;

Endpoints

Define API endpoints:

import {
  defineEndpoint,
  node,
  p,
  t,
  type InferParams,
  type InferOutputRow,
} from "@tinybirdco/sdk";

export const topPages = defineEndpoint("top_pages", {
  description: "Get the most visited pages",
  params: {
    start_date: p.dateTime(),
    end_date: p.dateTime(),
    limit: p.int32().optional(10),
  },
  nodes: [
    node({
      name: "aggregated",
      sql: `
        SELECT pathname, count() AS views
        FROM page_views
        WHERE timestamp >= {{DateTime(start_date)}}
          AND timestamp <= {{DateTime(end_date)}}
        GROUP BY pathname
        ORDER BY views DESC
        LIMIT {{Int32(limit, 10)}}
      `,
    }),
  ],
  output: {
    pathname: t.string(),
    views: t.uint64(),
  },
});

export type TopPagesParams = InferParams<typeof topPages>;
export type TopPagesOutput = InferOutputRow<typeof topPages>;

Pipes

Define internal pipes that are not exposed as API endpoints:

import { definePipe, node, p } from "@tinybirdco/sdk";

export const filteredEvents = definePipe("filtered_events", {
  description: "Filter events by date range",
  params: {
    start_date: p.dateTime(),
    end_date: p.dateTime(),
  },
  nodes: [
    node({
      name: "filtered",
      sql: `
        SELECT * FROM events
        WHERE timestamp >= {{DateTime(start_date)}}
          AND timestamp <= {{DateTime(end_date)}}
      `,
    }),
  ],
});

Materialized views

Use materialized views to populate derived datasources:

import {
  defineDatasource,
  defineMaterializedView,
  node,
  t,
  engine,
} from "@tinybirdco/sdk";

export const dailyStats = defineDatasource("daily_stats", {
  schema: {
    date: t.date(),
    pathname: t.string(),
    views: t.simpleAggregateFunction("sum", t.uint64()),
  },
  engine: engine.aggregatingMergeTree({
    sortingKey: ["date", "pathname"],
  }),
});

export const dailyStatsMv = defineMaterializedView("daily_stats_mv", {
  datasource: dailyStats,
  nodes: [
    node({
      name: "aggregate",
      sql: `
        SELECT toDate(timestamp) AS date, pathname, count() AS views
        FROM page_views
        GROUP BY date, pathname
      `,
    }),
  ],
});

Copy pipes

Use copy pipes for snapshots or scheduled exports:

import { defineCopyPipe, node } from "@tinybirdco/sdk";

export const dailySnapshot = defineCopyPipe("daily_snapshot", {
  datasource: dailyStats,
  schedule: "0 0 * * *",
  mode: "append",
  nodes: [
    node({
      name: "snapshot",
      sql: `SELECT * FROM daily_stats WHERE date = today() - 1`,
    }),
  ],
});

Sink pipes

Use sink pipes to publish query results to Kafka or S3:

import { defineSinkPipe, node } from "@tinybirdco/sdk";

export const kafkaEventsSink = defineSinkPipe("kafka_events_sink", {
  sink: {
    connection: eventsKafka,
    topic: "events_export",
    schedule: "@on-demand",
  },
  nodes: [
    node({
      name: "publish",
      sql: `SELECT timestamp, payload FROM kafka_events`,
    }),
  ],
});

export const s3EventsSink = defineSinkPipe("s3_events_sink", {
  sink: {
    connection: landingS3,
    bucketUri: "s3://my-bucket/exports/",
    fileTemplate: "events_{date}",
    format: "csv",
    schedule: "@once",
  },
  nodes: [
    node({
      name: "export",
      sql: `SELECT timestamp, session_id FROM s3_landing`,
    }),
  ],
});

Connections

Define connector connections and reference them in datasources:

import {
  defineGCSConnection,
  defineKafkaConnection,
  defineDatasource,
  engine,
  secret,
  t,
} from "@tinybirdco/sdk";

export const eventsKafka = defineKafkaConnection("events_kafka", {
  bootstrapServers: "kafka.example.com:9092",
  securityProtocol: "SASL_SSL",
  saslMechanism: "PLAIN",
  key: secret("KAFKA_KEY"),
  secret: secret("KAFKA_SECRET"),
});

export const landingGCS = defineGCSConnection("landing_gcs", {
  serviceAccountCredentialsJson: secret("GCS_SERVICE_ACCOUNT_CREDENTIALS_JSON"),
});

export const kafkaEvents = defineDatasource("kafka_events", {
  schema: {
    timestamp: t.dateTime(),
    payload: t.string(),
  },
  engine: engine.mergeTree({
    sortingKey: ["timestamp"],
  }),
  kafka: {
    connection: eventsKafka,
    topic: "events",
    groupId: "events-consumer",
  },
});

export const gcsLanding = defineDatasource("gcs_landing", {
  schema: {
    timestamp: t.dateTime(),
    session_id: t.string(),
  },
  engine: engine.mergeTree({
    sortingKey: ["timestamp"],
  }),
  gcs: {
    connection: landingGCS,
    bucketUri: "gs://my-gcs-bucket/events/*.csv",
    schedule: "@auto",
  },
});

Next steps

Updated