India's Open Data Ecosystem: Building with Government APIs

India's Open Data Ecosystem: Building with Government APIs

What is India's Open Data Ecosystem?

India has launched several government platforms providing open access to public data and digital services, creating opportunities for developers to build applications that serve millions.

data.gov.in: India's Central Data Portal

data.gov.in is the central repository for public datasets published by Indian government ministries and departments.

import requests
import pandas as pd
import json

# Get list of available datasets
def search_datasets(keyword): """Search data.gov.in for datasets""" api_url = 'https://data.gov.in/api/datastore_search' params = { 'resource_id': 'your_resource_id', 'q': keyword, 'limit': 100 } response = requests.get(api_url, params=params) datasets = response.json()['result']['records'] return datasets

# Example: Fetch Census 2021 data
def get_census_data(state_name): """Fetch Census 2021 population data by state""" api_url = 'https://data.gov.in/api/datastore_search' params = { 'resource_id': 'census-2021-population',  # Actual resource ID 'filters': {'state': state_name}, 'limit': 1000 } try: response = requests.get(api_url, params=params, timeout=10) response.raise_for_status() data = response.json()['result']['records'] df = pd.DataFrame(data) return df except requests.RequestException as e: print(f"Error fetching Census data: {e}") return None

# Example: Analyze population data
census_maharashtra = get_census_data('Maharashtra')
if census_maharashtra is not None: print(f"Districts in Maharashtra: {len(census_maharashtra)}") print(f"Total population: {census_maharashtra['population'].sum():,}") print(f"Literacy rate: {census_maharashtra['literacy_rate'].mean():.2f}%")

India Stack: Digital Infrastructure

1. Aadhaar (UIDAI)

Aadhaar is a 12-digit unique identification number issued to all Indian residents. The Unique Identification Authority (UIDAI) provides APIs for authentication and verification.

# Aadhaar Authentication API (simplified example)
# In production, would use UIDAI's official SDK

class AadhaarAuth: def __init__(self, api_key): self.api_key = api_key self.endpoint = 'https://api.uidai.gov.in/v1' def verify_aadhaar(self, aadhaar_number): """Verify Aadhaar number format""" # Aadhaar is 12 digits if len(aadhaar_number) == 12 and aadhaar_number.isdigit(): return True return False def authenticate_otp(self, aadhaar, otp): """Authenticate using OTP""" # In production: calls UIDAI API # Returns: success, demographic data pass def get_demographics(self, aadhaar_token): """Get demographic details (if authorized)""" # Returns: name, address, DOB, gender pass

# Usage
auth = AadhaarAuth(api_key='your_api_key')
is_valid = auth.verify_aadhaar('123456789012')
print(f"Aadhaar valid: {is_valid}")

2. UPI (Unified Payments Interface)

UPI enables instant money transfers between bank accounts. NPCI provides APIs for integration.

# UPI Payment Integration (simplified)
import requests
from datetime import datetime
import hashlib

class UPIPayment: def __init__(self, merchant_id, merchant_key): self.merchant_id = merchant_id self.merchant_key = merchant_key self.api_endpoint = 'https://api.npci.org.in/upi/v1' def initiate_payment(self, upi_id, amount, reference_id): """Initiate UPI payment""" payload = { 'merchantId': self.merchant_id, 'transactionId': reference_id, 'payeeVPA': upi_id, 'amount': amount, 'currency': 'INR', 'timestamp': datetime.utcnow().isoformat() } # Generate signature signature = self._generate_signature(payload) payload['signature'] = signature response = requests.post( f'{self.api_endpoint}/transactions', json=payload ) return response.json() def _generate_signature(self, payload): """Generate HMAC signature""" message = json.dumps(payload, sort_keys=True) return hashlib.sha256( (message + self.merchant_key).encode() ).hexdigest() def check_payment_status(self, transaction_id): """Check payment status""" params = { 'merchantId': self.merchant_id, 'transactionId': transaction_id } response = requests.get( f'{self.api_endpoint}/transactions/{transaction_id}', params=params ) return response.json()

# Usage
upi = UPIPayment(merchant_id='SHOP001', merchant_key='secret_key')
payment = upi.initiate_payment( upi_id='user@bankname', amount=500, reference_id='ORDER_12345'
)
print(f"Payment status: {payment['status']}")

3. DigiLocker

DigiLocker provides digital storage for official documents like licenses, certificates, and vehicle registration.

# DigiLocker Integration
import requests

class DigiLockerAPI: def __init__(self, client_id, client_secret): self.client_id = client_id self.client_secret = client_secret self.token_endpoint = 'https://digilocker.meity.gov.in/oauth/token' self.api_endpoint = 'https://digilocker.meity.gov.in/public/oauth2/1/issuer' def get_access_token(self): """Get OAuth token for accessing DigiLocker""" payload = { 'grant_type': 'client_credentials', 'client_id': self.client_id, 'client_secret': self.client_secret } response = requests.post(self.token_endpoint, json=payload) return response.json()['access_token'] def fetch_document(self, document_id, access_token): """Fetch a specific document""" headers = {'Authorization': f'Bearer {access_token}'} response = requests.get( f'{self.api_endpoint}/document/{document_id}', headers=headers ) return response.json() def list_documents(self, citizen_id, access_token): """List all documents for a citizen""" headers = {'Authorization': f'Bearer {access_token}'} params = {'citizen_id': citizen_id} response = requests.get( f'{self.api_endpoint}/documents', headers=headers, params=params ) documents = response.json()['documents'] return [ {'name': doc['docName'], 'issuer': doc['issuerName'], 'date': doc['issueDate']} for doc in documents ]

# Usage
digilocker = DigiLockerAPI(client_id='YOUR_CLIENT_ID', client_secret='YOUR_SECRET')
token = digilocker.get_access_token()
documents = digilocker.list_documents('citizen_uid', token)

for doc in documents: print(f"{doc['name']} - Issued by {doc['issuer']} on {doc['date']}")

e-NAM: Agricultural Market Platform

e-NAM provides real-time commodity prices from agricultural markets across India.

import requests
import pandas as pd

class ENAMMarketData: def __init__(self): self.api_base = 'https://api.enam.gov.in/v1' def get_commodity_prices(self, commodity_name, state=None): """Get prices for a commodity across markets""" endpoint = f'{self.api_base}/commodity/{commodity_name}' params = { 'limit': 100 } if state: params['state'] = state try: response = requests.get(endpoint, params=params) data = response.json() # Convert to DataFrame for analysis prices = [] for market in data['markets']: prices.append({ 'market': market['name'], 'state': market['state'], 'price_per_quintal': market['price'], 'highest': market['high'], 'lowest': market['low'], 'volume': market['volume_traded'] }) return pd.DataFrame(prices) except Exception as e: print(f"Error fetching e-NAM data: {e}") return None def analyze_price_trends(self, commodity, days=7): """Analyze price trends over time""" # Fetch historical data for commodity over N days prices_by_day = [] for day in range(days): # Would fetch real historical data pass return prices_by_day

# Usage
enam = ENAMMarketData()

# Get current wheat prices
wheat_prices = enam.get_commodity_prices('wheat')
print(wheat_prices[['market', 'state', 'price_per_quintal']].head())

# Find best market for wheat
best_market = wheat_prices.loc[wheat_prices['price_per_quintal'].idxmax()]
print(f"Best price for wheat: Rs {best_market['price_per_quintal']}/quintal in {best_market['market']}")

# Compare prices across states
state_avg = wheat_prices.groupby('state')['price_per_quintal'].agg(['mean', 'min', 'max'])
print(state_avg)

Open City Data: Municipal Services

Several Indian cities publish open data on public services, transportation, and infrastructure.

# Example: Bangalore Open Data API
class BangaloreOpenData: def __init__(self): self.api_base = 'https://data.bengaluru.gov.in/api' def get_traffic_data(self, area): """Get real-time traffic data""" endpoint = f'{self.api_base}/traffic' response = requests.get(endpoint, params={'area': area}) return response.json() def get_pothole_reports(self, limit=100): """Get reported potholes and road issues""" endpoint = f'{self.api_base}/infrastructure/potholes' response = requests.get(endpoint, params={'limit': limit}) data = response.json() return pd.DataFrame(data['reports']) def get_water_quality(self, ward_id): """Get water quality metrics""" endpoint = f'{self.api_base}/water/quality' response = requests.get(endpoint, params={'ward': ward_id}) return response.json() def get_public_transport(self, latitude, longitude, radius_km=1): """Find nearby buses and routes""" endpoint = f'{self.api_base}/transport/nearby' params = { 'lat': latitude, 'lon': longitude, 'radius': radius_km } response = requests.get(endpoint, params=params) return response.json()

# Usage
bangalore = BangaloreOpenData()

# Check traffic on a route
traffic = bangalore.get_traffic_data('MG Road')
print(f"Traffic density: {traffic['density']}%")

# Find nearby buses
buses = bangalore.get_public_transport(latitude=12.9716, longitude=77.5946)
for bus in buses['buses']: print(f"Route {bus['route']}: {bus['stop_distance']:.2f}m away")

Building an App with India Stack

class IndianEcommerceApp: """Example app using India Stack""" def __init__(self): self.aadhaar = AadhaarAuth() self.upi = UPIPayment() self.digilocker = DigiLockerAPI() def register_user(self, aadhaar_number, upi_id): """Register user with Aadhaar + UPI""" # Verify Aadhaar if not self.aadhaar.verify_aadhaar(aadhaar_number): return {'error': 'Invalid Aadhaar'} # Verify UPI ID if '@' not in upi_id: return {'error': 'Invalid UPI ID'} user = { 'aadhaar': aadhaar_number, 'upi': upi_id, 'verified': True, 'registration_time': datetime.utcnow().isoformat() } return user def process_payment(self, user, merchant_upi, amount): """Process payment via UPI""" payment_request = self.upi.initiate_payment( merchant_upi, amount, f"ORDER_{user['aadhaar']}_{int(time.time())}" ) return payment_request def verify_document(self, user, doc_type): """Verify user document from DigiLocker""" documents = self.digilocker.list_documents(user['aadhaar']) for doc in documents: if doc['type'] == doc_type: return {'verified': True, 'document': doc} return {'verified': False}

# Usage
app = IndianEcommerceApp()

# User registration
user = app.register_user('123456789012', 'user@upi')
print(f"User registered: {user}")

# Payment
payment = app.process_payment(user, 'merchant@upi', 5000)
print(f"Payment status: {payment['status']}")

# Document verification
doc = app.verify_document(user, 'driving_license')
print(f"Document verified: {doc['verified']}")

Practice Problems

1. Create an application that fetches Census data and visualizes population distribution
2. Build a commodity price comparison tool using e-NAM data for 5 Indian markets
3. Design an Aadhaar-based authentication system for a website
4. Create a UPI payment integration for an online store
5. Build a city services app using open municipal data APIs

Key Takeaways

• data.gov.in provides access to government datasets across multiple domains
• Aadhaar provides unique identification for over 1 billion Indians
• UPI enables seamless digital payments between any two bank accounts
• DigiLocker digitizes official documents for instant verification
• e-NAM brings agricultural market transparency and fair pricing
• Open city data enables civic tech applications for public benefit
• India Stack components can be combined for powerful applications
• These APIs are free and available to developers for public benefit