Python Code Examples
Complete, production-ready Python examples for ZhenRent API integration.
Prerequisites
pip install requests python-dotenv
Basic Setup
import os
import requests
from datetime import datetime, timedelta
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
API_KEY = os.getenv("ZHENRENT_API_KEY")
BASE_URL = "https://www.zhenrent.com/api/v1"
if not API_KEY:
raise ValueError("ZHENRENT_API_KEY environment variable not set")
# Default headers
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
Example 1: Create a Simple Task
def create_simple_task():
"""Create a basic photo verification task"""
response = requests.post(
f"{BASE_URL}/tasks",
headers=headers,
json={
"title": "Verify restaurant opening hours",
"description": "Please take a clear photo of the opening hours sign at the entrance",
"location": {
"latitude": 39.916527,
"longitude": 116.397128,
"address": "123 Main St, Beijing",
"max_distance_meters": 500
},
"budget_cents": 5000, # 50 RMB
"deadline_at": (datetime.now() + timedelta(days=7)).isoformat() + "Z",
"task_type": "photo",
"max_workers": 1
}
)
if response.status_code == 201:
task = response.json()
print(f"✅ Task created successfully!")
print(f" Task ID: {task['task_id']}")
print(f" Status: {task['status']}")
print(f" Total cost: {task['cost_breakdown']['total_cents']/100} RMB")
return task['task_id']
else:
error = response.json()['error']
print(f"❌ Error: {error['message']}")
return None
# Run it
task_id = create_simple_task()
Example 2: Complete Task Management Class
import uuid
import time
from typing import Optional, Dict, List
class ZhenRentClient:
"""Production-ready ZhenRent API client"""
def __init__(self, api_key: str, base_url: str = "https://www.zhenrent.com/api/v1"):
self.api_key = api_key
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def create_task(
self,
title: str,
description: str,
latitude: float,
longitude: float,
budget_cents: int,
deadline_days: int = 7,
task_type: str = "photo",
address: Optional[str] = None,
max_distance_meters: int = 500,
skills: Optional[List[str]] = None,
min_reputation: Optional[float] = None,
metadata: Optional[Dict] = None
) -> Optional[Dict]:
"""
Create a new task
Args:
title: Task title
description: Detailed instructions
latitude: Task latitude
longitude: Task longitude
budget_cents: Payment in cents (min 1000)
deadline_days: Days until deadline
task_type: Type of task
address: Human-readable address
max_distance_meters: Maximum worker distance
skills: Required worker skills
min_reputation: Minimum worker reputation
metadata: Custom metadata
Returns:
Task data if successful, None otherwise
"""
payload = {
"title": title,
"description": description,
"location": {
"latitude": latitude,
"longitude": longitude,
"max_distance_meters": max_distance_meters
},
"budget_cents": budget_cents,
"deadline_at": (datetime.now() + timedelta(days=deadline_days)).isoformat() + "Z",
"task_type": task_type,
"max_workers": 1,
"idempotency_key": str(uuid.uuid4())
}
if address:
payload["location"]["address"] = address
if skills or min_reputation:
payload["requirements"] = {}
if skills:
payload["requirements"]["skills"] = skills
if min_reputation:
payload["requirements"]["min_reputation"] = min_reputation
if metadata:
payload["metadata"] = metadata
try:
response = self.session.post(f"{self.base_url}/tasks", json=payload)
response.raise_for_status()
return response.json()
except requests.HTTPError as e:
print(f"HTTP Error: {e.response.status_code}")
print(f"Details: {e.response.json()}")
return None
except requests.RequestException as e:
print(f"Request Error: {e}")
return None
def get_task(self, task_id: str) -> Optional[Dict]:
"""Get task details"""
try:
response = self.session.get(f"{self.base_url}/tasks/{task_id}")
response.raise_for_status()
return response.json()
except requests.HTTPError:
return None
def list_tasks(
self,
status: Optional[str] = None,
limit: int = 20
) -> List[Dict]:
"""List all tasks"""
params = {"limit": limit}
if status:
params["status"] = status
try:
response = self.session.get(f"{self.base_url}/tasks", params=params)
response.raise_for_status()
return response.json()["data"]
except requests.HTTPError:
return []
def get_task_result(self, task_id: str) -> Optional[Dict]:
"""Get completed task result"""
try:
response = self.session.get(f"{self.base_url}/tasks/{task_id}/result")
response.raise_for_status()
return response.json()
except requests.HTTPError:
return None
def wait_for_completion(
self,
task_id: str,
timeout: int = 3600,
poll_interval: int = 30
) -> Optional[Dict]:
"""
Wait for task completion (polling)
Args:
task_id: Task ID to monitor
timeout: Maximum wait time in seconds
poll_interval: Seconds between checks
Returns:
Task data when completed, None if timeout
"""
start_time = time.time()
while time.time() - start_time < timeout:
task = self.get_task(task_id)
if not task:
print(f"❌ Task not found: {task_id}")
return None
status = task['status']
print(f"⏳ Task status: {status}")
if status == "completed":
print(f"✅ Task completed!")
return task
elif status in ["failed", "cancelled"]:
print(f"❌ Task {status}")
return task
time.sleep(poll_interval)
print(f"⏱️ Timeout after {timeout}s")
return None
def get_balance(self) -> Optional[Dict]:
"""Check account balance"""
try:
response = self.session.get(f"{self.base_url}/payment/balance")
response.raise_for_status()
return response.json()
except requests.HTTPError:
return None
# Usage example
client = ZhenRentClient(API_KEY)
# Create task
task = client.create_task(
title="Verify store opening hours",
description="Take a clear photo of the hour sign",
latitude=39.916527,
longitude=116.397128,
budget_cents=5000,
deadline_days=7,
skills=["photography"],
min_reputation=4.0,
metadata={"order_id": "ORD-12345"}
)
if task:
task_id = task['task_id']
print(f"Task created: {task_id}")
# Wait for completion
completed = client.wait_for_completion(task_id, timeout=7200)
if completed and completed['status'] == 'completed':
result = client.get_task_result(task_id)
print(f"Result: {result}")
Example 3: Batch Task Creation
def create_batch_tasks(locations: List[Dict]) -> List[str]:
"""
Create multiple tasks from a list of locations
Args:
locations: List of dicts with 'lat', 'lng', 'address', 'instructions'
Returns:
List of created task IDs
"""
client = ZhenRentClient(API_KEY)
task_ids = []
for i, loc in enumerate(locations, 1):
print(f"Creating task {i}/{len(locations)}...")
task = client.create_task(
title=f"Verify location: {loc['address']}",
description=loc.get('instructions', 'Take a clear photo of the location'),
latitude=loc['lat'],
longitude=loc['lng'],
budget_cents=5000,
address=loc['address'],
deadline_days=7
)
if task:
task_ids.append(task['task_id'])
print(f" ✅ Created: {task['task_id']}")
else:
print(f" ❌ Failed: {loc['address']}")
# Rate limiting: wait 1s between requests
time.sleep(1)
print(f"\nCreated {len(task_ids)}/{len(locations)} tasks")
return task_ids
# Example usage
locations = [
{
"lat": 39.916527,
"lng": 116.397128,
"address": "Tiananmen Square, Beijing",
"instructions": "Photo of the main entrance"
},
{
"lat": 39.9042,
"lng": 116.4074,
"address": "Jianguomenwai, Beijing",
"instructions": "Photo of the storefront"
},
{
"lat": 31.2304,
"lng": 121.4737,
"address": "Nanjing Road, Shanghai",
"instructions": "Photo of the building facade"
}
]
task_ids = create_batch_tasks(locations)
Example 4: Error Handling and Retries
import time
from functools import wraps
def retry_on_failure(max_retries=3, backoff_factor=2):
"""Decorator for retrying failed API calls"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except requests.HTTPError as e:
if e.response.status_code == 429:
# Rate limit - respect Retry-After header
retry_after = int(e.response.headers.get('Retry-After', 60))
print(f"Rate limited. Waiting {retry_after}s...")
time.sleep(retry_after)
continue
elif e.response.status_code >= 500:
# Server error - retry with backoff
if attempt < max_retries - 1:
wait_time = backoff_factor ** attempt
print(f"Server error. Retrying in {wait_time}s...")
time.sleep(wait_time)
continue
raise
except requests.RequestException as e:
# Network error - retry with backoff
if attempt < max_retries - 1:
wait_time = backoff_factor ** attempt
print(f"Network error. Retrying in {wait_time}s...")
time.sleep(wait_time)
continue
raise
raise Exception(f"Failed after {max_retries} attempts")
return wrapper
return decorator
class RobustZhenRentClient(ZhenRentClient):
"""ZhenRent client with robust error handling"""
@retry_on_failure(max_retries=3)
def create_task(self, *args, **kwargs):
"""Create task with automatic retries"""
result = super().create_task(*args, **kwargs)
if result is None:
raise requests.HTTPError("Task creation failed")
return result
def create_task_safe(self, *args, **kwargs) -> Optional[Dict]:
"""
Create task with comprehensive error handling
Returns task data or None on failure
"""
try:
return self.create_task(*args, **kwargs)
except requests.HTTPError as e:
if hasattr(e, 'response'):
error_data = e.response.json().get('error', {})
code = error_data.get('code')
if code == 'insufficient_balance':
balance = error_data.get('balance_cents', 0) / 100
required = error_data.get('required_cents', 0) / 100
print(f"❌ Insufficient balance: {balance} RMB (need {required} RMB)")
print(" Please recharge your account.")
elif code == 'validation_error':
print(f"❌ Validation error:")
for detail in error_data.get('details', []):
field = detail.get('loc', [])[-1]
msg = detail.get('msg')
print(f" - {field}: {msg}")
else:
print(f"❌ API Error [{code}]: {error_data.get('message')}")
else:
print(f"❌ HTTP Error: {e}")
except Exception as e:
print(f"❌ Unexpected error: {e}")
return None
# Usage
client = RobustZhenRentClient(API_KEY)
task = client.create_task_safe(
title="Test task",
description="Testing robust error handling",
latitude=39.916527,
longitude=116.397128,
budget_cents=5000
)
if task:
print(f"✅ Success: {task['task_id']}")
else:
print("❌ Failed to create task")
Example 5: CSV Import
import csv
def import_tasks_from_csv(csv_file: str) -> List[str]:
"""
Import tasks from CSV file
CSV format:
title,description,latitude,longitude,budget_rmb,address
"""
client = RobustZhenRentClient(API_KEY)
task_ids = []
with open(csv_file, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
task = client.create_task_safe(
title=row['title'],
description=row['description'],
latitude=float(row['latitude']),
longitude=float(row['longitude']),
budget_cents=int(float(row['budget_rmb']) * 100),
address=row.get('address'),
deadline_days=7
)
if task:
task_ids.append(task['task_id'])
print(f"✅ {row['title']}: {task['task_id']}")
else:
print(f"❌ Failed: {row['title']}")
time.sleep(1) # Rate limiting
return task_ids
# Usage
task_ids = import_tasks_from_csv('tasks.csv')
print(f"\nImported {len(task_ids)} tasks")
Example 6: Monitor Multiple Tasks
def monitor_tasks(task_ids: List[str], check_interval: int = 60):
"""Monitor multiple tasks until all complete"""
client = ZhenRentClient(API_KEY)
pending_tasks = set(task_ids)
completed_tasks = {}
print(f"Monitoring {len(task_ids)} tasks...")
while pending_tasks:
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] Checking {len(pending_tasks)} tasks...")
for task_id in list(pending_tasks):
task = client.get_task(task_id)
if not task:
print(f" ❌ {task_id[:8]}...: Not found")
pending_tasks.remove(task_id)
continue
status = task['status']
if status == 'completed':
print(f" ✅ {task_id[:8]}...: Completed")
completed_tasks[task_id] = task
pending_tasks.remove(task_id)
elif status in ['failed', 'cancelled']:
print(f" ❌ {task_id[:8]}...: {status}")
completed_tasks[task_id] = task
pending_tasks.remove(task_id)
else:
print(f" ⏳ {task_id[:8]}...: {status}")
if pending_tasks:
print(f"\nWaiting {check_interval}s before next check...")
time.sleep(check_interval)
print(f"\n✅ All tasks finished!")
print(f" Completed: {sum(1 for t in completed_tasks.values() if t['status'] == 'completed')}")
print(f" Failed: {sum(1 for t in completed_tasks.values() if t['status'] != 'completed')}")
return completed_tasks
# Usage
# First create some tasks
client = ZhenRentClient(API_KEY)
task_ids = []
for i in range(5):
task = client.create_task(
title=f"Test task {i+1}",
description="Testing monitoring",
latitude=39.916527 + i*0.01,
longitude=116.397128 + i*0.01,
budget_cents=5000
)
if task:
task_ids.append(task['task_id'])
# Monitor them
results = monitor_tasks(task_ids, check_interval=60)
Best Practices
1. Use Environment Variables
# .env file
ZHENRENT_API_KEY=zr_live_your_key
ZHENRENT_ENVIRONMENT=production
# Python code
from dotenv import load_dotenv
load_dotenv()
API_KEY = os.getenv("ZHENRENT_API_KEY")
2. Implement Logging
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# In your code
logger.info(f"Creating task: {title}")
logger.error(f"Failed to create task: {error}")
3. Use Type Hints
from typing import Optional, Dict, List
def create_task(
title: str,
budget_cents: int
) -> Optional[Dict]:
"""Type hints improve code clarity"""
pass
4. Validate Input
def validate_coordinates(lat: float, lng: float) -> bool:
"""Validate geographic coordinates"""
if not -90 <= lat <= 90:
raise ValueError(f"Invalid latitude: {lat}")
if not -180 <= lng <= 180:
raise ValueError(f"Invalid longitude: {lng}")
return True
def validate_budget(budget_cents: int) -> bool:
"""Validate task budget"""
if budget_cents < 1000:
raise ValueError(f"Budget must be at least 1000 cents (10 RMB)")
return True
Next Steps
- API Reference - Complete endpoint documentation
- Error Handling - Comprehensive error guide
- Authentication - Security best practices
- JavaScript Examples - Node.js code samples
- Go Examples - Golang implementations