Programming Guide & SDK Examples

This guide provides production-ready code examples for common Pidima API integration patterns. No official SDK is required — use any HTTP client.

Quick Reference

Language	Library	Auth Pattern
Python	requests	Session with Bearer token
JavaScript/Node.js	fetch / axios	Headers object
Shell/CLI	curl + jq	Environment variable
Java	HttpClient / WebClient	Interceptor

Domain Notice

All examples use https://staging.example.com as a placeholder base URL. Pidima can be deployed on any domain, IP, or air-gapped host. Replace with your actual instance address.

---

Python Integration

Complete Client Class

"""
Pidima API Client - Python
Requires: pip install requests
"""
import requests
from typing import Optional
from dataclasses import dataclass


class PidimaClient:
    """Minimal Pidima API client for programmatic access."""

    def __init__(self, base_url: str = "https://staging.example.com/api/v1.0"):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({"Content-Type": "application/json"})

    def login(self, email: str, password: str) -> dict:
        """Authenticate and store JWT token for subsequent requests."""
        resp = self.session.post(f"{self.base_url}/auth/login", json={
            "email": email,
            "password": password
        })
        resp.raise_for_status()
        data = resp.json()
        self.session.headers["Authorization"] = f"Bearer {data['token']}"
        return data

    # --- Projects ---

    def list_projects(self, account_id: str) -> list:
        """List all projects for an account."""
        resp = self.session.get(f"{self.base_url}/projects/accounts/{account_id}")
        resp.raise_for_status()
        return resp.json()

    def create_project(self, name: str, prefix: str, description: str,
                       account_id: str, admin_email: str = None) -> dict:
        """Create a new project."""
        resp = self.session.post(f"{self.base_url}/projects", json={
            "name": name,
            "prefix": prefix,
            "description": description,
            "accountId": account_id,
            "adminEmail": admin_email
        })
        resp.raise_for_status()
        return resp.json()

    # --- Requirements ---

    def create_requirement(self, project_id: str, description: str,
                           priority: str = "MEDIUM", req_type: str = "FUNCTIONAL",
                           status: str = "DRAFT", **kwargs) -> dict:
        """Create a single requirement."""
        payload = {
            "projectId": project_id,
            "description": description,
            "priority": priority,
            "type": req_type,
            "status": status,
            **kwargs
        }
        resp = self.session.post(f"{self.base_url}/requirements", json=payload)
        resp.raise_for_status()
        return resp.json()

    def list_requirements(self, project_id: str, page: int = 0,
                          size: int = 20) -> dict:
        """List requirements with pagination."""
        resp = self.session.get(
            f"{self.base_url}/requirements/projects/{project_id}/paginated",
            params={"page": page, "size": size}
        )
        resp.raise_for_status()
        return resp.json()

    def update_requirement(self, req_id: str, **fields) -> dict:
        """Update a requirement. Only pass fields you want to change."""
        resp = self.session.put(
            f"{self.base_url}/requirements/{req_id}", json=fields
        )
        resp.raise_for_status()
        return resp.json()

    # --- Test Cases ---

    def generate_testcases(self, project_id: str,
                           requirement_ids: list = None,
                           additional_context: str = None) -> dict:
        """Trigger AI test case generation (async)."""
        payload = {"projectId": project_id}
        if requirement_ids:
            payload["requirementIds"] = requirement_ids
        if additional_context:
            payload["additionalContext"] = additional_context
        resp = self.session.post(f"{self.base_url}/testcases/generate", json=payload)
        resp.raise_for_status()
        return resp.json()

    def list_testcases(self, project_id: str, page: int = 0,
                       size: int = 20) -> dict:
        """List test cases with pagination."""
        resp = self.session.get(
            f"{self.base_url}/testcases/projects/{project_id}/paginated",
            params={"page": page, "size": size}
        )
        resp.raise_for_status()
        return resp.json()

    # --- Traceability ---

    def create_traceability_link(self, requirement_id: str,
                                  testcase_ids: list) -> dict:
        """Link a requirement to test cases."""
        resp = self.session.post(f"{self.base_url}/traceability", json={
            "requirementId": requirement_id,
            "testcaseIds": testcase_ids
        })
        resp.raise_for_status()
        return resp.json()

    def get_traceability_matrix(self, project_id: str) -> list:
        """Get full traceability matrix for a project."""
        resp = self.session.get(
            f"{self.base_url}/traceability/projects/{project_id}"
        )
        resp.raise_for_status()
        return resp.json()

    # --- Search ---

    def search(self, query: str, project_id: str,
               types: list = None, limit: int = 20) -> dict:
        """Semantic search across requirements, test cases, architecture."""
        payload = {
            "query": query,
            "projectId": project_id,
            "limit": limit
        }
        if types:
            payload["types"] = types
        resp = self.session.post(f"{self.base_url}/search", json=payload)
        resp.raise_for_status()
        return resp.json()

    # --- Async Jobs ---

    def get_job_status(self, job_id: str) -> dict:
        """Check the status of an async job."""
        resp = self.session.get(f"{self.base_url}/jobs/{job_id}")
        resp.raise_for_status()
        return resp.json()

    def wait_for_job(self, job_id: str, timeout: int = 120,
                     poll_interval: int = 3) -> dict:
        """Poll a job until completion or timeout."""
        import time
        start = time.time()
        while time.time() - start < timeout:
            status = self.get_job_status(job_id)
            if status["status"] in ("COMPLETED", "FAILED", "CANCELLED"):
                return status
            time.sleep(poll_interval)
        raise TimeoutError(f"Job {job_id} did not complete within {timeout}s")

    # --- Export ---

    def export_requirements(self, project_id: str,
                            output_path: str, fmt: str = "excel") -> str:
        """Download requirements export as Excel or PDF."""
        resp = self.session.get(
            f"{self.base_url}/requirements/download/{project_id}",
            params={"format": fmt},
            stream=True
        )
        resp.raise_for_status()
        with open(output_path, "wb") as f:
            for chunk in resp.iter_content(chunk_size=8192):
                f.write(chunk)
        return output_path


# --- Usage Example ---

if __name__ == "__main__":
    client = PidimaClient("https://staging.example.com/api/v1.0")

    # Authenticate
    auth = client.login("engineer@acme.com", "password123")
    print(f"Logged in as: {auth['user']['name']}")

    # List projects
    projects = client.list_projects(account_id="your-account-uuid")
    project = projects[0]
    print(f"Working on: {project['name']}")

    # Create a requirement
    req = client.create_requirement(
        project_id=project["id"],
        description="The system shall detect obstacles within 150m range using radar sensors",
        priority="HIGH",
        name="REQ-RADAR-001",
        domain="Perception",
        customAttributes={"ASIL Level": "C"}
    )
    print(f"Created: {req['name']} (v{req['version']})")

    # Generate test cases from it
    job = client.generate_testcases(
        project_id=project["id"],
        requirement_ids=[req["id"]],
        additional_context="Include edge cases for adverse weather conditions"
    )
    print(f"Generation job: {job['jobId']}")

    # Wait for completion
    result = client.wait_for_job(job["jobId"])
    print(f"Generated {result['result']['generatedCount']} test cases")

    # Export everything
    client.export_requirements(project["id"], "output/requirements.xlsx")
    print("Export complete!")

---

JavaScript / Node.js Integration

Complete Client Module

/**
 * Pidima API Client - Node.js
 * No external dependencies required (uses native fetch)
 * For Node.js < 18, install: npm install node-fetch
 */

class PidimaClient {
  constructor(baseUrl = 'https://staging.example.com/api/v1.0') {
    this.baseUrl = baseUrl.replace(/\/$/, '');
    this.token = null;
  }

  async #request(method, path, { body, params, stream } = {}) {
    const url = new URL(`${this.baseUrl}${path}`);
    if (params) {
      Object.entries(params).forEach(([k, v]) => {
        if (v !== undefined) url.searchParams.set(k, v);
      });
    }

    const headers = { 'Content-Type': 'application/json' };
    if (this.token) headers['Authorization'] = `Bearer ${this.token}`;

    const resp = await fetch(url.toString(), {
      method,
      headers,
      body: body ? JSON.stringify(body) : undefined
    });

    if (!resp.ok) {
      const error = await resp.json().catch(() => ({ message: resp.statusText }));
      throw new Error(`API Error ${resp.status}: ${error.message}`);
    }

    if (stream) return resp;
    const text = await resp.text();
    return text ? JSON.parse(text) : null;
  }

  // --- Auth ---
  async login(email, password) {
    const data = await this.#request('POST', '/auth/login', {
      body: { email, password }
    });
    this.token = data.token;
    return data;
  }

  // --- Projects ---
  async listProjects(accountId) {
    return this.#request('GET', `/projects/accounts/${accountId}`);
  }

  async createProject({ name, prefix, description, accountId }) {
    return this.#request('POST', '/projects', {
      body: { name, prefix, description, accountId }
    });
  }

  // --- Requirements ---
  async createRequirement({ projectId, description, priority = 'MEDIUM',
                            type = 'FUNCTIONAL', status = 'DRAFT', ...rest }) {
    return this.#request('POST', '/requirements', {
      body: { projectId, description, priority, type, status, ...rest }
    });
  }

  async listRequirements(projectId, { page = 0, size = 20 } = {}) {
    return this.#request('GET', `/requirements/projects/${projectId}/paginated`, {
      params: { page, size }
    });
  }

  async updateRequirement(id, fields) {
    return this.#request('PUT', `/requirements/${id}`, { body: fields });
  }

  // --- Test Cases ---
  async generateTestcases({ projectId, requirementIds, additionalContext }) {
    return this.#request('POST', '/testcases/generate', {
      body: { projectId, requirementIds, additionalContext }
    });
  }

  async listTestcases(projectId, { page = 0, size = 20 } = {}) {
    return this.#request('GET', `/testcases/projects/${projectId}/paginated`, {
      params: { page, size }
    });
  }

  // --- Traceability ---
  async createTraceabilityLink(requirementId, testcaseIds) {
    return this.#request('POST', '/traceability', {
      body: { requirementId, testcaseIds }
    });
  }

  async getTraceabilityMatrix(projectId) {
    return this.#request('GET', `/traceability/projects/${projectId}`);
  }

  // --- Search ---
  async search(query, projectId, { types, limit = 20 } = {}) {
    return this.#request('POST', '/search', {
      body: { query, projectId, types, limit }
    });
  }

  // --- Jobs ---
  async getJobStatus(jobId) {
    return this.#request('GET', `/jobs/${jobId}`);
  }

  async waitForJob(jobId, { timeout = 120000, interval = 3000 } = {}) {
    const start = Date.now();
    while (Date.now() - start < timeout) {
      const status = await this.getJobStatus(jobId);
      if (['COMPLETED', 'FAILED', 'CANCELLED'].includes(status.status)) {
        return status;
      }
      await new Promise(r => setTimeout(r, interval));
    }
    throw new Error(`Job ${jobId} timed out after ${timeout}ms`);
  }
}

// --- Usage Example ---

async function main() {
  const client = new PidimaClient('https://staging.example.com/api/v1.0');

  // Authenticate
  const auth = await client.login('engineer@acme.com', 'password123');
  console.log(`Logged in as: ${auth.user.name}`);

  // List projects
  const projects = await client.listProjects('your-account-uuid');
  const project = projects[0];
  console.log(`Working on: ${project.name}`);

  // Create requirement
  const req = await client.createRequirement({
    projectId: project.id,
    name: 'REQ-RADAR-001',
    description: 'The system shall detect obstacles within 150m using radar',
    priority: 'HIGH',
    domain: 'Perception',
    customAttributes: { 'ASIL Level': 'C' }
  });
  console.log(`Created: ${req.name} (v${req.version})`);

  // Generate test cases
  const job = await client.generateTestcases({
    projectId: project.id,
    requirementIds: [req.id],
    additionalContext: 'Include adverse weather edge cases'
  });
  console.log(`Job started: ${job.jobId}`);

  // Wait for result
  const result = await client.waitForJob(job.jobId);
  console.log(`Generated ${result.result.generatedCount} test cases`);

  // Semantic search
  const searchResults = await client.search(
    'radar obstacle detection failure',
    project.id,
    { types: ['REQUIREMENT', 'TESTCASE'] }
  );
  console.log(`Found ${searchResults.results.length} matches`);
}

main().catch(console.error);

---

Shell / CLI Scripts

Environment Setup

#!/bin/bash
# pidima-env.sh - Source this file before running API commands
export PIDIMA_URL="https://staging.example.com/api/v1.0"
export PIDIMA_EMAIL="engineer@acme.com"
export PIDIMA_PASSWORD="password123"

# Login and export token
export PIDIMA_TOKEN=$(curl -s -X POST "$PIDIMA_URL/auth/login" \
  -H "Content-Type: application/json" \
  -d "{\"email\": \"$PIDIMA_EMAIL\", \"password\": \"$PIDIMA_PASSWORD\"}" \
  | jq -r '.token')

echo "✓ Authenticated. Token set in PIDIMA_TOKEN"

Common Operations Script

#!/bin/bash
# pidima-ops.sh - Common operations helper
# Usage: source pidima-env.sh && ./pidima-ops.sh <command> [args...]

API="$PIDIMA_URL"
AUTH="Authorization: Bearer $PIDIMA_TOKEN"

case "$1" in
  list-projects)
    curl -s "$API/projects/accounts/$2" -H "$AUTH" | jq '.[] | {id, name, prefix}'
    ;;

  create-requirement)
    curl -s -X POST "$API/requirements" \
      -H "$AUTH" -H "Content-Type: application/json" \
      -d "{
        \"projectId\": \"$2\",
        \"name\": \"$3\",
        \"description\": \"$4\",
        \"priority\": \"${5:-MEDIUM}\",
        \"type\": \"FUNCTIONAL\",
        \"status\": \"DRAFT\"
      }" | jq '{id, name, version}'
    ;;

  generate-tests)
    curl -s -X POST "$API/testcases/generate" \
      -H "$AUTH" -H "Content-Type: application/json" \
      -d "{
        \"projectId\": \"$2\",
        \"requirementIds\": [\"$3\"]
      }" | jq '{jobId, status}'
    ;;

  job-status)
    curl -s "$API/jobs/$2" -H "$AUTH" | jq '{status, progress}'
    ;;

  search)
    curl -s -X POST "$API/search" \
      -H "$AUTH" -H "Content-Type: application/json" \
      -d "{
        \"query\": \"$3\",
        \"projectId\": \"$2\",
        \"types\": [\"REQUIREMENT\", \"TESTCASE\"],
        \"limit\": 10
      }" | jq '.results[] | {type, name, score}'
    ;;

  export-reqs)
    curl -s -o "requirements-export.xlsx" \
      "$API/requirements/download/$2?format=excel" -H "$AUTH"
    echo "✓ Exported to requirements-export.xlsx"
    ;;

  *)
    echo "Commands: list-projects, create-requirement, generate-tests, job-status, search, export-reqs"
    ;;
esac

---

CI/CD Integration Example

GitHub Actions: Auto-generate Tests on PR

# .github/workflows/pidima-tests.yml
name: Generate Test Cases from Requirements
on:
  push:
    paths: ['requirements/**']

jobs:
  generate-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Login to Pidima
        id: auth
        run: |
          TOKEN=$(curl -s -X POST "${{ vars.PIDIMA_URL }}/auth/login" \
            -H "Content-Type: application/json" \
            -d '{"email": "${{ secrets.PIDIMA_EMAIL }}", "password": "${{ secrets.PIDIMA_PASSWORD }}"}' \
            | jq -r '.token')
          echo "token=$TOKEN" >> $GITHUB_OUTPUT

      - name: Trigger Test Generation
        run: |
          JOB_ID=$(curl -s -X POST "${{ vars.PIDIMA_URL }}/testcases/generate" \
            -H "Authorization: Bearer ${{ steps.auth.outputs.token }}" \
            -H "Content-Type: application/json" \
            -d '{"projectId": "${{ vars.PIDIMA_PROJECT_ID }}"}' \
            | jq -r '.jobId')

          # Poll until done
          for i in $(seq 1 40); do
            STATUS=$(curl -s "${{ vars.PIDIMA_URL }}/jobs/$JOB_ID" \
              -H "Authorization: Bearer ${{ steps.auth.outputs.token }}" \
              | jq -r '.status')
            echo "Job status: $STATUS"
            if [ "$STATUS" = "COMPLETED" ]; then exit 0; fi
            if [ "$STATUS" = "FAILED" ]; then exit 1; fi
            sleep 5
          done
          echo "Timeout" && exit 1

---

Common Integration Patterns

Pattern 1: Bulk Import Requirements from CSV

import csv
from pidima_client import PidimaClient  # Use the class above

client = PidimaClient()
client.login("engineer@acme.com", "password123")

with open("requirements.csv") as f:
    reader = csv.DictReader(f)
    for row in reader:
        req = client.create_requirement(
            project_id="your-project-uuid",
            name=row["ID"],
            description=row["Description"],
            priority=row.get("Priority", "MEDIUM").upper(),
            req_type=row.get("Type", "FUNCTIONAL").upper(),
            domain=row.get("Domain", ""),
            customAttributes={
                "Source": row.get("Source", ""),
                "Rationale": row.get("Rationale", "")
            }
        )
        print(f"  Created: {req['name']}")

Pattern 2: Automated Traceability Check

def check_traceability_coverage(client, project_id):
    """Report requirements missing test case coverage."""
    reqs = client.list_requirements(project_id, page=0, size=1000)
    matrix = client.get_traceability_matrix(project_id)

    covered_req_ids = {link["requirementId"] for link in matrix}
    uncovered = [
        r for r in reqs["content"]
        if r["id"] not in covered_req_ids and r["status"] == "APPROVED"
    ]

    coverage = 1 - (len(uncovered) / max(reqs["totalElements"], 1))
    print(f"Traceability Coverage: {coverage:.0%}")
    print(f"Uncovered approved requirements: {len(uncovered)}")
    for r in uncovered:
        print(f"  ⚠ {r['name']}: {r['description'][:80]}...")

    return coverage, uncovered

Pattern 3: Impact Analysis Workflow

def monitor_document_changes(client, project_id, document_id):
    """Upload new document version and check impact."""
    # Trigger impact analysis
    result = client.session.post(
        f"{client.base_url}/impact-analysis/trigger",
        json={"projectId": project_id, "documentId": document_id}
    ).json()

    # Wait for analysis
    job = client.wait_for_job(result["jobId"])

    # Get impacted items
    report = client.session.get(
        f"{client.base_url}/impact-analysis/reports/{result['reportId']}"
    ).json()

    print(f"Impact Analysis Complete:")
    print(f"  Requirements affected: {report['impactedRequirements']}")
    print(f"  Test cases affected: {report['impactedTestcases']}")
    print(f"  Architecture affected: {report['impactedArchitectures']}")
    return report

---

Rate Limiting & Best Practices

Guideline	Recommendation
Batch operations	Use batch endpoints (/requirements/batch) for bulk creates
Polling interval	Poll async jobs every 3–5 seconds, not faster
Page size	Use size=50–100 for bulk reads, size=10–20 for UI
Token refresh	Re-authenticate before token expires (24h)
Error handling	Retry on 5xx with exponential backoff (max 3 retries)
File uploads	Stream large files, respect 50MB limit
Concurrent requests	Limit to 10 parallel requests per user

---

Swagger / OpenAPI (on the backend)

When the Pidima backend application is running, an interactive API explorer is available:

https://<your-host>/swagger-ui/index.html

info

This is served by the pidima-app Java application itself — not by the documentation site. The backend must be running and accessible. The JSON spec is at /v3/api-docs/pidima-api.