Track orders, products, and origins across your supply chain using Coolset APIs.
Manage your end-to-end supply chain with complete visibility from source to delivery. Track orders, monitor product flows, and ensure traceability.
Create and track purchase and sale orders
Monitor product movements through the value chain
Document and verify product origins
Map complete supply chain relationships
# 1. Create a purchase order
curl -X POST https://developers-scranton.coolset.com/api/orders/ \
-H "Authorization: Bearer YOUR_API_TOKEN" \
-H "Content-Type: application/json" \
-d '{
"identifier": "PO-2024-001",
"type": "purchase",
"supplier": "Supplier Ltd",
"status": "pending"
}'
# 2. Add products to order
curl -X POST https://developers-scranton.coolset.com/api/orders/123/order-items/ \
-H "Authorization: Bearer YOUR_API_TOKEN" \
-H "Content-Type: application/json" \
-d '{
"product_id": 456,
"quantity": 1000,
"unit": "kg"
}'
# 3. Link to origin
curl -X POST https://developers-scranton.coolset.com/api/origins/ \
-H "Authorization: Bearer YOUR_API_TOKEN" \
-H "Content-Type: application/json" \
-d '{
"name": "Brazilian Farm A",
"country": "BR",
"region": "Minas Gerais",
"geolocation": {
"latitude": -19.9167,
"longitude": -43.9345
}
}'
# 4. Update order status
curl -X PATCH https://developers-scranton.coolset.com/api/orders/123/ \
-H "Authorization: Bearer YOUR_API_TOKEN" \
-H "Content-Type: application/json" \
-d '{
"status": "in_transit"
}'import requests
from datetime import datetime
class CoolsetSupplyChain:
def __init__(self, api_token):
self.base_url = "https://developers-scranton.coolset.com/api"
self.headers = {
"Authorization": f"Bearer {api_token}",
"Content-Type": "application/json"
}
def create_purchase_order(self, supplier, items):
"""Create a new purchase order with items"""
# Create order
order = requests.post(
f"{self.base_url}/orders/",
headers=self.headers,
json={
"identifier": f"PO-{datetime.now().strftime('%Y%m%d-%H%M')}",
"type": "purchase",
"supplier": supplier,
"status": "pending"
}
).json()
# Add items
for item in items:
requests.post(
f"{self.base_url}/orders/{order['id']}/order-items/",
headers=self.headers,
json=item
)
return order
def track_order_status(self, order_id):
"""Get current order status and items"""
order = requests.get(
f"{self.base_url}/orders/{order_id}/",
headers=self.headers
).json()
items = requests.get(
f"{self.base_url}/orders/{order_id}/order-items/",
headers=self.headers
).json()
return {
"order": order,
"items": items['results']
}
def update_order_status(self, order_id, new_status):
"""Update order status"""
return requests.patch(
f"{self.base_url}/orders/{order_id}/",
headers=self.headers,
json={"status": new_status}
).json()
def map_value_chain(self, product_id):
"""Get complete value chain for a product"""
value_chains = requests.get(
f"{self.base_url}/value-chains/",
headers=self.headers,
params={"product_id": product_id}
).json()
return value_chains['results']
# Usage
sc = CoolsetSupplyChain("YOUR_API_TOKEN")
# Create order
order = sc.create_purchase_order(
supplier="Coffee Co-op Brazil",
items=[
{
"product_id": 456,
"quantity": 5000,
"unit": "kg",
"price_per_unit": 4.50
}
]
)
print(f"Created order: {order['identifier']}")
# Track order
status = sc.track_order_status(order['id'])
print(f"Current status: {status['order']['status']}")
# Update status
sc.update_order_status(order['id'], "in_transit")curl -X POST https://developers-scranton.coolset.com/api/products/ \
-H "Authorization: Bearer YOUR_API_TOKEN" \
-H "Content-Type: application/json" \
-d '{
"name": "Arabica Coffee Beans",
"sku": "COFFEE-AR-001",
"category": "raw_materials",
"unit_of_measure": "kg",
"origin_country": "BR"
}'curl -X PATCH https://developers-scranton.coolset.com/api/products/456/ \
-H "Authorization: Bearer YOUR_API_TOKEN" \
-H "Content-Type: application/json" \
-d '{
"origins": [123, 124, 125]
}'Document the complete journey from source to customer:
curl -X POST https://developers-scranton.coolset.com/api/value-chains/ \
-H "Authorization: Bearer YOUR_API_TOKEN" \
-H "Content-Type: application/json" \
-d '{
"name": "Coffee Supply Chain - Brazil to EU",
"stages": [
{
"name": "Farming",
"location": "Brazilian Farm",
"type": "production"
},
{
"name": "Processing",
"location": "Local Mill",
"type": "processing"
},
{
"name": "Export",
"location": "Santos Port",
"type": "logistics"
},
{
"name": "Import",
"location": "Hamburg Port",
"type": "logistics"
},
{
"name": "Roasting",
"location": "EU Facility",
"type": "processing"
}
]
}'Get insights into your supply chain:
curl -X GET "https://developers-scranton.coolset.com/api/orders/statistics/?start_date=2024-01-01&end_date=2024-12-31" \
-H "Authorization: Bearer YOUR_API_TOKEN"Response:
{
"total_orders": 245,
"total_value": 1250000.00,
"by_status": {
"pending": 12,
"in_transit": 45,
"completed": 188
},
"by_type": {
"purchase": 150,
"sale": 95
},
"top_suppliers": [
{
"name": "Supplier A",
"order_count": 45
}
]
}Implement automatic traceability tracking:
class TraceabilityManager:
def __init__(self, api_token):
self.api = CoolsetSupplyChain(api_token)
def trace_product_journey(self, product_id):
"""Trace complete product journey"""
# Get all orders containing this product
orders = self.get_product_orders(product_id)
# Get value chains
value_chains = self.api.map_value_chain(product_id)
# Get origins
origins = self.get_product_origins(product_id)
return {
"product_id": product_id,
"orders": orders,
"value_chains": value_chains,
"origins": origins,
"traceability_score": self.calculate_traceability_score(
orders, value_chains, origins
)
}
def calculate_traceability_score(self, orders, chains, origins):
"""Calculate traceability completeness"""
score = 0
if orders:
score += 0.4 # Has order data
if chains:
score += 0.3 # Has value chain mapping
if origins:
score += 0.3 # Has origin data
return score
# Usage
tm = TraceabilityManager("YOUR_API_TOKEN")
trace = tm.trace_product_journey(456)
print(f"Traceability Score: {trace['traceability_score'] * 100}%")Always use clear, unique identifiers:
def generate_order_id(order_type, date):
"""Generate structured order IDs"""
prefix = "PO" if order_type == "purchase" else "SO"
timestamp = date.strftime("%Y%m%d-%H%M%S")
return f"{prefix}-{timestamp}"Log every status change for audit trail:
def update_with_logging(order_id, new_status, reason):
"""Update status with logging"""
# Get current status
order = api.get_order(order_id)
old_status = order['status']
# Update status
api.update_order_status(order_id, new_status)
# Log change
log_status_change(
order_id=order_id,
old_status=old_status,
new_status=new_status,
reason=reason,
timestamp=datetime.now()
)Keep detailed origin information:
origin_data = {
"name": "Farm Name",
"country": "BR",
"region": "State/Province",
"geolocation": {
"latitude": -19.9167,
"longitude": -43.9345,
"precision": "4_decimal_places" # ~11m accuracy
},
"certifications": [
"Organic",
"Fair Trade",
"Rainforest Alliance"
],
"contact": {
"name": "Farm Manager",
"email": "manager@farm.com"
}
}Receive real-time updates on order changes:
@app.route('/webhooks/coolset/order-updated', methods=['POST'])
def handle_order_update():
data = request.json
order_id = data['order_id']
new_status = data['status']
# Update your internal systems
update_internal_order_status(order_id, new_status)
# Trigger notifications
if new_status == 'completed':
notify_warehouse_team(order_id)
return {'status': 'processed'}Process multiple orders efficiently:
def process_daily_orders():
"""Process all pending orders"""
# Get pending orders
orders = requests.get(
f"{API_URL}/orders/",
headers=headers,
params={"status": "pending", "limit": 100}
).json()
for order in orders['results']:
try:
# Process order
process_order(order)
# Update status
update_order_status(order['id'], 'processing')
except Exception as e:
log_error(f"Failed to process order {order['id']}: {e}")POST /orders/- Supply Chain APIGET /orders/{id}/- Supply Chain APIPATCH /orders/{id}/- Supply Chain APIPOST /products/- Supply Chain APIPOST /origins/- Supply Chain APIPOST /value-chains/- Supply Chain API