Skip to main content

Code Examples

Complete code examples for integrating with the Heimdall API in various programming languages.

JavaScript / TypeScript​

REST API Client​

class HeimdallClient {
  constructor(baseUrl, apiToken) {
    this.baseUrl = baseUrl;
    this.apiToken = apiToken;
  }

  async request(endpoint, options = {}) {
    const url = `${this.baseUrl}${endpoint}`;
    const headers = {
      'Content-Type': 'application/json',
      ...options.headers,
    };

    if (this.apiToken) {
      headers['Authorization'] = `Bearer ${this.apiToken}`;
    }

    const response = await fetch(url, {
      ...options,
      headers,
    });

    if (!response.ok) {
      const error = await response.json();
      throw new Error(error.message || 'API request failed');
    }

    return response.json();
  }

  // Health check
  async health() {
    return this.request('/health');
  }

  // Get current GPS location
  async getCurrentGps() {
    return this.request('/v1/gps/current');
  }

  // List GPS data with pagination
  async listGps(page = 1, limit = 10) {
    return this.request(`/v1/gps?page=${page}&limit=${limit}`);
  }

  // Get GPS by ID
  async getGpsById(id) {
    return this.request(`/v1/gps/${id}`);
  }

  // Create GPS data (admin only)
  async createGps(data) {
    return this.request('/v1/gps', {
      method: 'POST',
      body: JSON.stringify(data),
    });
  }

  // GraphQL query
  async graphql(query, variables = {}) {
    return this.request('/v1/graphql', {
      method: 'POST',
      body: JSON.stringify({ query, variables }),
    });
  }
}

// Usage
const client = new HeimdallClient('https://api.elcto.com', 'YOUR_TOKEN');

// Get current GPS
const current = await client.getCurrentGps();
console.log(current.data);

// Create new GPS data
const newGps = await client.createGps({
  latitude: 51.5074,
  longitude: -0.1278,
  altitude: 11.0,
  timestamp: Date.now() / 1000,
  speed: 5.5,
});
console.log(newGps.data);

TypeScript with Type Safety​

interface GpsData {
  id: string;
  latitude: number;
  longitude: number;
  altitude: number;
  timestamp: number;
  speed: number;
  createdAt: string;
}

interface PaginatedResponse<T> {
  data: T[];
  pagination: {
    page: number;
    limit: number;
    total: number;
    total_pages: number;
  };
  _links: {
    self: string;
    first?: string;
    prev?: string;
    next?: string;
    last?: string;
  };
}

interface ResourceResponse<T> {
  data: T;
  _links: {
    self: string;
  };
}

class HeimdallClient {
  constructor(
    private baseUrl: string,
    private apiToken?: string
  ) {}

  private async request<T>(
    endpoint: string,
    options: RequestInit = {}
  ): Promise<T> {
    const url = `${this.baseUrl}${endpoint}`;
    const headers: HeadersInit = {
      'Content-Type': 'application/json',
      ...options.headers,
    };

    if (this.apiToken) {
      headers['Authorization'] = `Bearer ${this.apiToken}`;
    }

    const response = await fetch(url, {
      ...options,
      headers,
    });

    if (!response.ok) {
      const error = await response.json();
      throw new Error(error.message || 'API request failed');
    }

    return response.json();
  }

  async getCurrentGps(): Promise<ResourceResponse<GpsData>> {
    return this.request('/v1/gps/current');
  }

  async listGps(
    page = 1,
    limit = 10
  ): Promise<PaginatedResponse<GpsData>> {
    return this.request(`/v1/gps?page=${page}&limit=${limit}`);
  }

  async createGps(
    data: Omit<GpsData, 'id' | 'createdAt'>
  ): Promise<ResourceResponse<GpsData>> {
    return this.request('/v1/gps', {
      method: 'POST',
      body: JSON.stringify(data),
    });
  }
}

WebSocket Client​

class GpsWebSocket {
  constructor(url) {
    this.url = url;
    this.ws = null;
    this.listeners = new Map();
    this.connect();
  }

  connect() {
    this.ws = new WebSocket(this.url);

    this.ws.onopen = () => {
      console.log('Connected');
      this.subscribe('gps');
    };

    this.ws.onmessage = (event) => {
      const message = JSON.parse(event.data);
      this.handleMessage(message);
    };

    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error);
    };

    this.ws.onclose = () => {
      console.log('Disconnected, reconnecting...');
      setTimeout(() => this.connect(), 3000);
    };
  }

  handleMessage(message) {
    if (message.type === 'Ping') {
      this.send({ type: 'Pong' });
      return;
    }

    if (message.type === 'Message') {
      const data = JSON.parse(message.data);
      this.emit(message.channel, data);
    }
  }

  subscribe(channel) {
    this.send({
      type: 'Subscribe',
      channel,
    });
  }

  send(data) {
    if (this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify(data));
    }
  }

  on(event, callback) {
    if (!this.listeners.has(event)) {
      this.listeners.set(event, []);
    }
    this.listeners.get(event).push(callback);
  }

  emit(event, data) {
    const callbacks = this.listeners.get(event) || [];
    callbacks.forEach(callback => callback(data));
  }
}

// Usage
const ws = new GpsWebSocket('wss://api.elcto.com/v1/ws');

ws.on('gps', (data) => {
  console.log('GPS update:', data);
});

Python​

REST API Client​

import requests
from typing import Optional, Dict, Any, List

class HeimdallClient:
    def __init__(self, base_url: str, api_token: Optional[str] = None):
        self.base_url = base_url
        self.api_token = api_token
        self.session = requests.Session()

        if api_token:
            self.session.headers.update({
                'Authorization': f'Bearer {api_token}',
                'Content-Type': 'application/json'
            })

    def _request(
        self,
        method: str,
        endpoint: str,
        **kwargs
    ) -> Dict[str, Any]:
        url = f"{self.base_url}{endpoint}"
        response = self.session.request(method, url, **kwargs)
        response.raise_for_status()
        return response.json()

    def health(self) -> Dict[str, Any]:
        """Check API health status"""
        return self._request('GET', '/health')

    def get_current_gps(self) -> Dict[str, Any]:
        """Get current GPS location"""
        return self._request('GET', '/v1/gps/current')

    def list_gps(
        self,
        page: int = 1,
        limit: int = 10
    ) -> Dict[str, Any]:
        """List GPS data with pagination"""
        return self._request(
            'GET',
            f'/v1/gps?page={page}&limit={limit}'
        )

    def get_gps_by_id(self, gps_id: str) -> Dict[str, Any]:
        """Get GPS data by ID"""
        return self._request('GET', f'/v1/gps/{gps_id}')

    def create_gps(
        self,
        latitude: float,
        longitude: float,
        altitude: float,
        timestamp: int,
        speed: float
    ) -> Dict[str, Any]:
        """Create new GPS data (admin only)"""
        return self._request(
            'POST',
            '/v1/gps',
            json={
                'latitude': latitude,
                'longitude': longitude,
                'altitude': altitude,
                'timestamp': timestamp,
                'speed': speed
            }
        )

    def graphql(
        self,
        query: str,
        variables: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """Execute GraphQL query"""
        return self._request(
            'POST',
            '/v1/graphql',
            json={
                'query': query,
                'variables': variables or {}
            }
        )

# Usage
client = HeimdallClient('https://api.elcto.com', 'YOUR_TOKEN')

# Get current GPS
current = client.get_current_gps()
print(current['data'])

# List GPS data
gps_list = client.list_gps(page=1, limit=20)
for gps in gps_list['data']:
    print(f"Location: {gps['latitude']}, {gps['longitude']}")

# Create GPS data
new_gps = client.create_gps(
    latitude=51.5074,
    longitude=-0.1278,
    altitude=11.0,
    timestamp=int(time.time()),
    speed=5.5
)
print(f"Created: {new_gps['data']['id']}")

WebSocket Client​

import asyncio
import json
import websockets

class GpsWebSocket:
    def __init__(self, url: str):
        self.url = url
        self.ws = None
        self.running = False

    async def connect(self):
        """Connect to WebSocket"""
        self.ws = await websockets.connect(self.url)
        self.running = True
        print("✅ Connected")

        # Subscribe to GPS updates
        await self.subscribe('gps')

        # Start message handler
        await self.handle_messages()

    async def subscribe(self, channel: str):
        """Subscribe to a channel"""
        await self.send({
            'type': 'Subscribe',
            'channel': channel
        })

    async def send(self, data: dict):
        """Send message to server"""
        if self.ws:
            await self.ws.send(json.dumps(data))

    async def handle_messages(self):
        """Handle incoming messages"""
        try:
            async for message in self.ws:
                data = json.loads(message)
                await self.handle_message(data)
        except websockets.exceptions.ConnectionClosed:
            print("❌ Connection closed")
            self.running = False

    async def handle_message(self, message: dict):
        """Handle individual message"""
        msg_type = message.get('type')

        if msg_type == 'Ping':
            await self.send({'type': 'Pong'})
        elif msg_type == 'Message':
            channel = message.get('channel')
            data = json.loads(message.get('data', '{}'))
            print(f"📍 {channel}: {data}")
        elif msg_type == 'Error':
            print(f"❌ Error: {message.get('message')}")

# Usage
async def main():
    ws = GpsWebSocket('wss://api.elcto.com/v1/ws')
    await ws.connect()

asyncio.run(main())

Rust​

REST API Client​

use reqwest::{Client, Response};
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
pub struct GpsData {
    pub id: String,
    pub latitude: f64,
    pub longitude: f64,
    pub altitude: f64,
    pub timestamp: i64,
    pub speed: f64,
    pub created_at: String,
}

#[derive(Debug, Deserialize)]
pub struct ResourceResponse<T> {
    pub data: T,
    pub _links: Links,
}

#[derive(Debug, Deserialize)]
pub struct Links {
    #[serde(rename = "self")]
    pub self_link: String,
}

pub struct HeimdallClient {
    base_url: String,
    client: Client,
}

impl HeimdallClient {
    pub fn new(base_url: String, api_token: Option<String>) -> Self {
        let mut headers = reqwest::header::HeaderMap::new();
        headers.insert(
            reqwest::header::CONTENT_TYPE,
            reqwest::header::HeaderValue::from_static("application/json"),
        );

        if let Some(token) = api_token {
            headers.insert(
                reqwest::header::AUTHORIZATION,
                reqwest::header::HeaderValue::from_str(&format!("Bearer {}", token))
                    .expect("Invalid token"),
            );
        }

        let client = Client::builder()
            .default_headers(headers)
            .build()
            .expect("Failed to build client");

        Self { base_url, client }
    }

    pub async fn get_current_gps(&self) -> Result<ResourceResponse<GpsData>, reqwest::Error> {
        let url = format!("{}/v1/gps/current", self.base_url);
        self.client.get(&url).send().await?.json().await
    }

    pub async fn create_gps(
        &self,
        latitude: f64,
        longitude: f64,
        altitude: f64,
        timestamp: i64,
        speed: f64,
    ) -> Result<ResourceResponse<GpsData>, reqwest::Error> {
        let url = format!("{}/v1/gps", self.base_url);
        let body = serde_json::json!({
            "latitude": latitude,
            "longitude": longitude,
            "altitude": altitude,
            "timestamp": timestamp,
            "speed": speed,
        });

        self.client.post(&url).json(&body).send().await?.json().await
    }
}

// Usage
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = HeimdallClient::new(
        "https://api.elcto.com".to_string(),
        Some("YOUR_TOKEN".to_string()),
    );

    let current = client.get_current_gps().await?;
    println!("Current GPS: {:?}", current.data);

    Ok(())
}

Go​

REST API Client​

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "net/http"
)

type GpsData struct {
    ID        string  `json:"id"`
    Latitude  float64 `json:"latitude"`
    Longitude float64 `json:"longitude"`
    Altitude  float64 `json:"altitude"`
    Timestamp int64   `json:"timestamp"`
    Speed     float64 `json:"speed"`
    CreatedAt string  `json:"createdAt"`
}

type ResourceResponse struct {
    Data   GpsData          `json:"data"`
    Links  map[string]string `json:"_links"`
}

type HeimdallClient struct {
    BaseURL  string
    APIToken string
    Client   *http.Client
}

func NewClient(baseURL, apiToken string) *HeimdallClient {
    return &HeimdallClient{
        BaseURL:  baseURL,
        APIToken: apiToken,
        Client:   &http.Client{},
    }
}

func (c *HeimdallClient) doRequest(method, endpoint string, body interface{}) (*http.Response, error) {
    url := c.BaseURL + endpoint

    var reqBody []byte
    if body != nil {
        var err error
        reqBody, err = json.Marshal(body)
        if err != nil {
            return nil, err
        }
    }

    req, err := http.NewRequest(method, url, bytes.NewBuffer(reqBody))
    if err != nil {
        return nil, err
    }

    req.Header.Set("Content-Type", "application/json")
    if c.APIToken != "" {
        req.Header.Set("Authorization", "Bearer "+c.APIToken)
    }

    return c.Client.Do(req)
}

func (c *HeimdallClient) GetCurrentGps() (*ResourceResponse, error) {
    resp, err := c.doRequest("GET", "/v1/gps/current", nil)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()

    var result ResourceResponse
    if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
        return nil, err
    }

    return &result, nil
}

func (c *HeimdallClient) CreateGps(latitude, longitude, altitude float64, timestamp int64, speed float64) (*ResourceResponse, error) {
    body := map[string]interface{}{
        "latitude":  latitude,
        "longitude": longitude,
        "altitude":  altitude,
        "timestamp": timestamp,
        "speed":     speed,
    }

    resp, err := c.doRequest("POST", "/v1/gps", body)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()

    var result ResourceResponse
    if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
        return nil, err
    }

    return &result, nil
}

func main() {
    client := NewClient("https://api.elcto.com", "YOUR_TOKEN")

    current, err := client.GetCurrentGps()
    if err != nil {
        panic(err)
    }

    fmt.Printf("Current GPS: %+v\n", current.Data)
}

Next Steps​