Moose APIs

Viewing:

Overview

The APIs module provides standalone HTTP endpoints for data ingestion and analytics. Unlike other modules of the MooseStack, APIs are meant to be paired with other MooseStack modules like OLAP tables and streams.

Core Capabilities

Ingest

Ingest New Data

Create HTTP POST endpoints for data ingestion with validation and routing

Analytics

Expose Analytics

Create HTTP GET endpoints for reading data from your OLAP database with query parameters

Trigger

Trigger Workflows

Create HTTP POST endpoints to trigger workflows and other processes

Basic Examples

Export Required

Ensure your API is correctly exported from your app/index.ts file.

Learn more about export pattern: local development / hosted.

Ingestion API

IngestApi.ts

import { IngestApi } from "@514labs/moose-lib";
 
interface UserEvent {
  id: string;
  userId: string;
  timestamp: Date;
  eventType: string;
}
 
// Create a standalone ingestion API
export const userEventsApi = new IngestApi<UserEvent>("user-events", {
  destination: eventStream
});

IngestApi.py

from moose_lib import IngestApi
from pydantic import BaseModel
from datetime import datetime
 
class UserEvent(BaseModel):
    id: str
    user_id: str
    timestamp: datetime
    event_type: str
 
# Create a standalone ingestion API
user_events_api = IngestApi[UserEvent]("user-events", IngestConfig(destination=event_stream))
 
# No export needed - Python modules are automatically discovered

Analytics API

AnalyticsApi.ts

import { Api } from "@514labs/moose-lib";
 
interface Params {
  userId: string;
  limit: number;
}
 
interface ResultData {
  id: string;
  name: string;
  email: string;
}
 
// Create a standalone analytics API
export const userDataApi = new Api<Params, ResultData[]>("user-data", 
  async ({ userId, limit }, { client, sql }) => {
    // Query external service or custom logic
    return [
      { id: userId, name: "John Doe", email: "john@example.com" }
    ];
  }
);

AnalyticsApi.py

from moose_lib import Api, MooseClient
from pydantic import BaseModel
 
class Params(BaseModel):
    user_id: str
    limit: int
 
class ResultData(BaseModel):
    id: str
    name: str
    email: str
 
def query_function(client: MooseClient, params: QueryParams) -> list[UserData]:
    # Query external service or custom logic using parameter binding
    query = "SELECT * FROM user_data WHERE user_id = {user_id} LIMIT {limit}"
    return client.query.execute(query, {"user_id": params.user_id, "limit": params.limit})
 
user_data_api = Api[Params, ResultData]("get-data", query_function)
 
# No export needed - Python modules are automatically discovered

Cancelling Running Workflows

Ingest New Data