import { Key, OlapTable, Stream, IngestApi, ConsumptionApi } from "@514labs/moose-lib";
 
interface DataModel {
  primaryKey: Key<string>;
  name: string;
}
// Create a ClickHouse table
export const clickhouseTable = new OlapTable<DataModel>("TableName");
 
// Create a Redpanda streaming topic
export const redpandaTopic = new Stream<DataModel>("TopicName", {
  destination: clickhouseTable,
});
 
// Create an ingest API endpoint
export const ingestApi = new IngestApi<DataModel>("post-api-route", {
  destination: redpandaTopic,
});
 
// Create consumption API endpoint
interface QueryParams {
  limit?: number;
}
export const consumptionApi = new ConsumptionApi<QueryParams, DataModel>("get-api-route", {
  async handler({limit = 10}: QueryParams, {client, sql}): {
    const result = await client.query.execute(sql`SELECT * FROM ${clickhouseTable} LIMIT ${limit}`);
    return await result.json();
  }
});

from moose_lib import Key, OlapTable, Stream, StreamConfig, IngestApi, IngestApiConfig, ConsumptionApi
from pydantic import BaseModel
 
class DataModel(BaseModel):
    primary_key: Key[str]
    name: str
 
# Create a ClickHouse table
clickhouse_table = OlapTable[DataModel]("TableName")
 
# Create a Redpanda streaming topic
redpanda_topic = Stream[DataModel]("TopicName", StreamConfig(
    destination=clickhouse_table,
))
 
# Create an ingest API endpoint
ingest_api = IngestApi[DataModel]("post-api-route", IngestApiConfig(
    destination=redpanda_topic,
))
 
# Create a consumption API endpoint
class QueryParams(BaseModel):
    limit: int = 10
 
def handler(client, params: QueryParams):
    return client.query.execute("SELECT * FROM {table: Identifier} LIMIT {limit: Int32}", {
        "table": clickhouse_table.name,
        "limit": params.limit,
    })
 
consumption_api = ConsumptionApi[RequestParams, DataModel]("get-api-route", query_function=handler)

Kafka Config     (300 lines of YAML)
+ ClickHouse     (Manual schema management)  
+ dbt            (SQL-only transformations)
+ Airflow        (Workflow orchestration)
+ Custom APIs    (Boilerplate code)
+ ...

const pipeline = new IngestPipeline<DataModel>("pipeline", {
  ingest: true, // HTTP API w/ validation
  stream: true, // Redpanda/Kafka topic
  table: true, // ClickHouse table
});