JavaScript/Node.js SDK
A complete JavaScript SDK for managing parking whitelists:Copy
class ArqqinParkingAPI {
constructor(apiKey, baseURL = 'https://apistg.arqq.in/api') {
this.apiKey = apiKey;
this.baseURL = baseURL;
this.client = axios.create({
baseURL,
headers: {
'X-API-Key': apiKey,
'Content-Type': 'application/json'
}
});
}
// Register a new company (requires API key)
static async registerCompany(companyData, apiKey, baseURL = 'https://apistg.arqq.in/api') {
try {
const client = axios.create({
baseURL,
headers: {
'X-API-Key': apiKey,
'Content-Type': 'application/json'
}
});
const response = await client.post('/register/company', companyData);
return response.data;
} catch (error) {
console.error('Error registering company:', error.response?.data || error.message);
throw error;
}
}
// List all companies
async getCompanies() {
try {
const response = await this.client.get('/companies');
return response.data;
} catch (error) {
console.error('Error fetching companies:', error.response?.data || error.message);
throw error;
}
}
// List all accessible locations
async getLocations() {
try {
const response = await this.client.get('/locations');
return response.data;
} catch (error) {
console.error('Error fetching locations:', error.response?.data || error.message);
throw error;
}
}
// Add a new location
async addLocation(locationData) {
try {
const response = await this.client.post('/locations', locationData);
return response.data;
} catch (error) {
console.error('Error adding location:', error.response?.data || error.message);
throw error;
}
}
// Update an existing location
async updateLocation(locationId, locationData) {
try {
const response = await this.client.put(`/locations/${locationId}`, locationData);
return response.data;
} catch (error) {
console.error('Error updating location:', error.response?.data || error.message);
throw error;
}
}
// Get bulk location status for all company locations
async getBulkLocationStatus() {
try {
const response = await this.client.get('/locations/status');
return response.data;
} catch (error) {
console.error('Error fetching bulk location status:', error.response?.data || error.message);
throw error;
}
}
// Get location details
async getLocation(locationId) {
try {
const response = await this.client.get(`/locations/${locationId}`);
return response.data;
} catch (error) {
console.error('Error fetching location:', error.response?.data || error.message);
throw error;
}
}
// Get location status
async getLocationStatus(locationId) {
try {
const response = await this.client.get(`/locations/${locationId}/status`);
return response.data;
} catch (error) {
console.error('Error fetching location status:', error.response?.data || error.message);
throw error;
}
}
// List whitelist vehicles with optional filters
async listWhitelist(locationId, filters = {}) {
try {
const params = new URLSearchParams();
if (filters.referenceId) params.append('referenceId', filters.referenceId);
if (filters.plateType) params.append('plateType', filters.plateType);
if (filters.plateCode) params.append('plateCode', filters.plateCode);
if (filters.plateNumber) params.append('plateNumber', filters.plateNumber);
if (filters.limit) params.append('limit', String(filters.limit));
const response = await this.client.get(`/locations/${locationId}/whitelist?${params}`);
return response.data;
} catch (error) {
console.error('Error listing whitelist:', error.response?.data || error.message);
throw error;
}
}
// Add a vehicle to the whitelist
async addWhitelistVehicle(locationId, whitelistEntry) {
try {
const response = await this.client.post(`/locations/${locationId}/whitelist`, whitelistEntry);
return response.data;
} catch (error) {
console.error('Error adding whitelist vehicle:', error.response?.data || error.message);
throw error;
}
}
// Update a whitelist vehicle
async updateWhitelistVehicle(locationId, whitelistId, updates) {
try {
const response = await this.client.put(`/locations/${locationId}/whitelist/${whitelistId}`, updates);
return response.data;
} catch (error) {
console.error('Error updating whitelist vehicle:', error.response?.data || error.message);
throw error;
}
}
// Delete a whitelist vehicle
async deleteWhitelistVehicle(locationId, whitelistId) {
try {
const response = await this.client.delete(`/locations/${locationId}/whitelist/${whitelistId}`);
return response.data;
} catch (error) {
console.error('Error deleting whitelist vehicle:', error.response?.data || error.message);
throw error;
}
}
// Health check
async healthCheck() {
try {
const response = await this.client.get('/health');
return response.data;
} catch (error) {
console.error('Error checking health:', error.response?.data || error.message);
throw error;
}
}
}
// Usage examples
// 1. Register a new company (requires API key)
ArqqinParkingAPI.registerCompany({
name: "Downtown Property Management",
slug: "downtown-pm",
externalCompanyId: "EXT_12345",
description: "Premium property management services",
address: "123 Business Street, Dubai",
phone: "+1234567890",
email: "[email protected]",
website: "https://downtownpm.com",
logo: "https://storage.example.com/logo.png",
settings: {}
}, 'YOUR_API_KEY')
.then(result => console.log('Company registered:', result))
.catch(error => console.error('Registration error:', error));
// 2. Use authenticated API instance
const api = new ArqqinParkingAPI('your-api-key');
// 3. List companies
api.getCompanies()
.then(result => {
console.log('Companies:', result.data);
result.data.forEach(company => {
console.log(`- ${company.name} (${company.slug})`);
});
})
.catch(error => console.error('Error:', error));
// Add a new location
api.addLocation({
name: "Sunset Towers",
slug: "sunset-towers",
address: "456 Sunset Boulevard, Dubai Marina",
city: "Dubai",
country: "UAE",
coordinates: { lat: 25.0772, lng: 55.1409 }
})
.then(result => console.log('Location added:', result))
.catch(error => console.error('Error:', error));
// Update an existing location
api.updateLocation('location_123', {
name: "Sunset Towers Updated",
address: "456 Sunset Boulevard (Renovated), Dubai Marina",
city: "Dubai",
country: "UAE",
coordinates: { lat: 25.0773, lng: 55.1410 }
})
.then(result => console.log('Location updated:', result))
.catch(error => console.error('Error:', error));
// Get bulk status for all locations
api.getBulkLocationStatus()
.then(result => console.log('Bulk status:', result))
.catch(error => console.error('Error:', error));
// List whitelist vehicles for a reference
api.listWhitelist('location_123', { referenceId: 'REF_001' })
.then(result => console.log('Whitelist:', result))
.catch(error => console.error('Error:', error));
Python SDK
A comprehensive Python SDK for parking management:Copy
import requests
from typing import Optional, Dict, Any, List
class ArqqinParkingAPI:
def __init__(self, api_key: str, base_url: str = 'https://apistg.arqq.in/api'):
self.api_key = api_key
self.base_url = base_url
self.headers = {
'X-API-Key': api_key,
'Content-Type': 'application/json'
}
def _make_request(self, method: str, endpoint: str, data: Optional[Dict] = None, params: Optional[Dict] = None) -> Dict[str, Any]:
url = f"{self.base_url}{endpoint}"
try:
if method.upper() == 'GET':
response = requests.get(url, headers=self.headers, params=params)
elif method.upper() == 'POST':
response = requests.post(url, headers=self.headers, json=data)
elif method.upper() == 'PUT':
response = requests.put(url, headers=self.headers, json=data)
elif method.upper() == 'DELETE':
response = requests.delete(url, headers=self.headers)
else:
raise ValueError(f"Unsupported HTTP method: {method}")
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"API request failed: {e}")
if hasattr(e, 'response') and e.response is not None:
print(f"Response: {e.response.text}")
raise
@staticmethod
def register_company(company_data: Dict[str, Any], base_url: str = 'https://apistg.arqq.in/api') -> Dict[str, Any]:
"""Register a new company (no authentication required)"""
headers = {'Content-Type': 'application/json'}
try:
response = requests.post(f"{base_url}/register/company", headers=headers, json=company_data)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"Company registration failed: {e}")
if hasattr(e, 'response') and e.response is not None:
print(f"Response: {e.response.text}")
raise
def get_locations(self) -> Dict[str, Any]:
"""Get all accessible locations"""
return self._make_request('GET', '/locations')
def add_location(self, location_data: Dict[str, Any]) -> Dict[str, Any]:
"""Add a new location"""
return self._make_request('POST', '/locations', data=location_data)
def update_location(self, location_id: str, location_data: Dict[str, Any]) -> Dict[str, Any]:
"""Update an existing location"""
return self._make_request('PUT', f'/locations/{location_id}', data=location_data)
def get_bulk_location_status(self) -> Dict[str, Any]:
"""Get bulk location status for all company locations"""
return self._make_request('GET', '/locations/status')
def get_location(self, location_id: str) -> Dict[str, Any]:
"""Get location details"""
return self._make_request('GET', f'/locations/{location_id}')
def get_location_status(self, location_id: str) -> Dict[str, Any]:
"""Get location status and statistics"""
return self._make_request('GET', f'/locations/{location_id}/status')
def list_whitelist(self, location_id: str,
reference_id: Optional[str] = None,
plate_type: Optional[str] = None,
plate_code: Optional[str] = None,
plate_number: Optional[str] = None,
limit: Optional[int] = None) -> Dict[str, Any]:
"""List whitelist vehicles for a location with optional filters"""
params = {}
if reference_id:
params['referenceId'] = reference_id
if plate_type:
params['plateType'] = plate_type
if plate_code:
params['plateCode'] = plate_code
if plate_number:
params['plateNumber'] = plate_number
if limit:
params['limit'] = limit
return self._make_request('GET', f'/locations/{location_id}/whitelist', params=params)
def add_whitelist_vehicle(self, location_id: str, entry: Dict[str, Any]) -> Dict[str, Any]:
"""Add a vehicle to the whitelist"""
return self._make_request('POST', f'/locations/{location_id}/whitelist', data=entry)
def update_whitelist_vehicle(self, location_id: str, whitelist_id: str, updates: Dict[str, Any]) -> Dict[str, Any]:
"""Update a vehicle in the whitelist"""
return self._make_request('PUT', f'/locations/{location_id}/whitelist/{whitelist_id}', data=updates)
def delete_whitelist_vehicle(self, location_id: str, whitelist_id: str) -> Dict[str, Any]:
"""Delete a vehicle from the whitelist"""
return self._make_request('DELETE', f'/locations/{location_id}/whitelist/{whitelist_id}')
def health_check(self) -> Dict[str, Any]:
"""Check API health status"""
return self._make_request('GET', '/health')
@staticmethod
def register_company(company_data: Dict[str, Any], api_key: str, base_url: str = 'https://apistg.arqq.in/api') -> Dict[str, Any]:
"""Register a new company (requires API key)"""
headers = {
'X-API-Key': api_key,
'Content-Type': 'application/json'
}
try:
response = requests.post(f"{base_url}/register/company", headers=headers, json=company_data)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"Registration failed: {e}")
if hasattr(e, 'response') and e.response is not None:
print(f"Response: {e.response.text}")
raise
def get_companies(self) -> Dict[str, Any]:
"""Get list of companies"""
return self._make_request('GET', '/companies')
# Usage example
api = ArqqinParkingAPI('your-api-key')
try:
# Test connection
health = api.health_check()
print("API Health:", health)
# Get companies
companies = api.get_companies()
print("Companies:", companies['data'])
# Get locations
locations = api.get_locations()
print("Available locations:", locations['data'])
# List whitelist for a reference
result = api.list_whitelist('location_123', reference_id='REF_001')
print("Whitelist:", result)
except Exception as e:
print(f"Error: {e}")
cURL Examples
Complete cURL examples for all API endpoints:Copy
curl -X GET "https://apistg.arqq.in/api/health" \
-H "X-API-Key: YOUR_API_KEY" \
-H "Content-Type: application/json"
Real-World Integration Examples
Property Management System Integration
Resident Portal Integration
Resident Portal Integration
Integrate with a resident portal to manage parking:
Copy
class ResidentParkingManager {
constructor(api) {
this.api = api;
this.locationId = 'your-location-id';
}
async addResidentVehicle(residentId, plateData) {
const vehicleData = {
referenceId: `RESIDENT_${residentId}`,
...plateData,
note: `Vehicle for resident ${residentId}`
};
try {
const result = await this.api.addWhitelistVehicle(this.locationId, vehicleData);
return { success: true, data: result.data };
} catch (error) {
if (error.response?.status === 409) {
return { success: false, error: 'Vehicle already exists' };
}
throw error;
}
}
async getResidentVehicles(residentId) {
return await this.api.listWhitelist(
this.locationId,
{ referenceId: `RESIDENT_${residentId}` }
);
}
async removeResidentVehicle(whitelistId) {
return await this.api.deleteWhitelistVehicle(this.locationId, whitelistId);
}
}
Bulk Import from CSV
Bulk Import from CSV
Import vehicles from a CSV file:
Copy
import csv
import asyncio
from typing import List, Dict
class BulkVehicleImporter:
def __init__(self, api: ArqqinParkingAPI):
self.api = api
async def import_from_csv(self, csv_file_path: str, location_id: str):
vehicles = []
with open(csv_file_path, 'r') as file:
reader = csv.DictReader(file)
for row in reader:
vehicle_data = {
'referenceId': row['reference_id'],
'plateType': row['plate_type'],
'plateCode': row['plate_code'],
'plateNumber': row['plate_number'],
'vehicleType': row.get('vehicle_type'),
'note': row.get('note')
}
vehicles.append(vehicle_data)
# Process in batches to avoid rate limits
batch_size = 10
results = []
for i in range(0, len(vehicles), batch_size):
batch = vehicles[i:i + batch_size]
tasks = []
for vehicle in batch:
task = self.api.add_whitelist_vehicle(location_id, vehicle)
tasks.append(task)
batch_results = await asyncio.gather(*tasks, return_exceptions=True)
results.extend(batch_results)
# Wait between batches
if i + batch_size < len(vehicles):
await asyncio.sleep(1)
return results
Webhook Integration
Webhook Handler for Parking Events
Webhook Handler for Parking Events
Handle parking events with webhooks:
Copy
const express = require('express');
const app = express();
app.use(express.json());
// Webhook endpoint for parking events
app.post('/webhook/parking-events', (req, res) => {
const { event, data } = req.body;
switch (event) {
case 'vehicle_entered':
handleVehicleEntry(data);
break;
case 'vehicle_exited':
handleVehicleExit(data);
break;
case 'unauthorized_vehicle':
handleUnauthorizedVehicle(data);
break;
default:
console.log('Unknown event type:', event);
}
res.status(200).json({ received: true });
});
function handleVehicleEntry(data) {
console.log(`Vehicle ${data.plateType} ${data.plateCode} ${data.plateNumber} entered at ${data.timestamp}`);
// Update your system with entry information
}
function handleVehicleExit(data) {
console.log(`Vehicle ${data.plateType} ${data.plateCode} ${data.plateNumber} exited at ${data.timestamp}`);
// Update your system with exit information
}
function handleUnauthorizedVehicle(data) {
console.log(`Unauthorized vehicle attempt: ${data.plateType} ${data.plateCode} ${data.plateNumber}`);
// Handle unauthorized access attempt
}
Error Handling Patterns
Comprehensive Error Handling
Comprehensive Error Handling
Implement robust error handling:
Copy
class APIErrorHandler {
static handleError(error) {
if (error.response) {
const { status, data } = error.response;
switch (status) {
case 400:
return { type: 'validation', message: data.error?.message || 'Invalid request' };
case 401:
return { type: 'auth', message: 'Invalid API key' };
case 403:
return { type: 'permission', message: 'Access denied' };
case 404:
return { type: 'not_found', message: 'Resource not found' };
case 409:
return { type: 'conflict', message: 'Vehicle already exists' };
case 429:
return { type: 'rate_limit', message: 'Rate limit exceeded' };
case 500:
return { type: 'server', message: 'Internal server error' };
default:
return { type: 'unknown', message: 'An unexpected error occurred' };
}
} else if (error.request) {
return { type: 'network', message: 'Network error - unable to reach API' };
} else {
return { type: 'client', message: error.message };
}
}
}
// Usage
try {
const result = await api.addWhitelistVehicle(locationId, vehicleData);
} catch (error) {
const errorInfo = APIErrorHandler.handleError(error);
console.error(`${errorInfo.type}: ${errorInfo.message}`);
}
Testing Your Integration
Unit Tests for API Integration
Unit Tests for API Integration
Test your integration with unit tests:
Copy
import unittest
from unittest.mock import Mock, patch
from arqqin_api import ArqqinParkingAPI
class TestArqqinAPI(unittest.TestCase):
def setUp(self):
self.api = ArqqinParkingAPI('test-api-key')
@patch('requests.get')
def test_get_locations(self, mock_get):
mock_response = Mock()
mock_response.json.return_value = {
'success': True,
'data': [{'id': 'location_123', 'name': 'Test Location'}]
}
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
result = self.api.get_locations()
self.assertTrue(result['success'])
self.assertEqual(len(result['data']), 1)
self.assertEqual(result['data'][0]['id'], 'location_123')
@patch('requests.post')
def test_add_vehicle(self, mock_post):
mock_response = Mock()
mock_response.json.return_value = {
'success': True,
'data': {'id': 'whitelist_789', 'plateNumber': '12345'}
}
mock_response.raise_for_status.return_value = None
mock_post.return_value = mock_response
vehicle_data = {
'referenceId': 'RESIDENT_001',
'plateType': 'DXB',
'plateCode': 'White',
'plateNumber': '12345'
}
result = self.api.add_whitelist_vehicle('location_123', vehicle_data)
self.assertTrue(result['success'])
self.assertEqual(result['data']['plateNumber'], '12345')
if __name__ == '__main__':
unittest.main()

