Python Examples¶
Python client examples using the requests library.
Requirements¶
Quick Start¶
import requests
import base64
import os
# Configuration
CLIENT_ID = os.environ['CLIENT_ID']
CLIENT_SECRET = os.environ['CLIENT_SECRET']
API_KEY = os.environ['API_KEY']
BASE_URL = 'https://sandbox.lms.api.fsri.org/api/partner/v1'
TOKEN_URL = 'https://fsri-partner-api-sandbox-357795418386.auth.us-west-2.amazoncognito.com/oauth2/token'
# Get access token
def get_token():
credentials = base64.b64encode(
f"{CLIENT_ID}:{CLIENT_SECRET}".encode()
).decode()
response = requests.post(
TOKEN_URL,
headers={
'Content-Type': 'application/x-www-form-urlencoded',
'Authorization': f'Basic {credentials}'
},
data='grant_type=client_credentials&scope=partner-api/read partner-api/write'
)
response.raise_for_status()
return response.json()['access_token']
# Make API request
token = get_token()
response = requests.get(
f'{BASE_URL}/ping',
headers={'Authorization': f'Bearer {token}', 'x-api-key': API_KEY}
)
print(response.json())
Complete Client Class¶
"""
FSA LMS Partner API Client
Usage:
from lms_client import LMSClient
client = LMSClient(
client_id=os.environ['CLIENT_ID'],
client_secret=os.environ['CLIENT_SECRET'],
api_key=os.environ['API_KEY']
)
# Register a user
user = client.register_user('jane@example.com', 'Jane', 'Doe')
# List courses
courses = client.get_courses(user['uuid'])
"""
import requests
import base64
import time
from datetime import datetime, timedelta, timezone
from typing import Optional, Dict, Any, List
class LMSClient:
"""Client for the FSA LMS Partner API."""
def __init__(
self,
client_id: str,
client_secret: str,
api_key: str,
base_url: str = 'https://sandbox.lms.api.fsri.org/api/partner/v1',
token_url: str = 'https://fsri-partner-api-sandbox-357795418386.auth.us-west-2.amazoncognito.com/oauth2/token'
):
self.client_id = client_id
self.client_secret = client_secret
self.api_key = api_key # sent as x-api-key on every request
self.base_url = base_url.rstrip('/')
self.token_url = token_url
self._token: Optional[str] = None
self._token_expires_at: Optional[datetime] = None
def _get_token(self) -> str:
"""Get or refresh the access token."""
if self._token and self._token_expires_at:
if datetime.now() < self._token_expires_at:
return self._token
# client_credentials has no refresh tokens — "refreshing" = minting anew
credentials = base64.b64encode(
f"{self.client_id}:{self.client_secret}".encode()
).decode()
response = requests.post(
self.token_url,
headers={
'Content-Type': 'application/x-www-form-urlencoded',
'Authorization': f'Basic {credentials}'
},
data='grant_type=client_credentials&scope=partner-api/read partner-api/write'
)
response.raise_for_status()
data = response.json()
self._token = data['access_token']
# Re-mint 5 minutes before expiry
self._token_expires_at = datetime.now() + timedelta(
seconds=data['expires_in'] - 300
)
return self._token
def _request(
self,
method: str,
endpoint: str,
user_uuid: Optional[str] = None,
json_data: Optional[Dict] = None,
params: Optional[Dict] = None,
retry_count: int = 3
) -> requests.Response:
"""Make an authenticated API request with retry logic."""
headers = {
'Authorization': f'Bearer {self._get_token()}',
'x-api-key': self.api_key,
}
if user_uuid:
headers['X-User-ID'] = user_uuid
if json_data:
# PATCH uses JSON Merge Patch (RFC 7396); everything else is plain JSON
headers['Content-Type'] = (
'application/merge-patch+json' if method.upper() == 'PATCH' else 'application/json'
)
url = f"{self.base_url}{endpoint}"
for attempt in range(retry_count):
response = requests.request(
method,
url,
headers=headers,
json=json_data,
params=params
)
# Handle token expiration
if response.status_code == 401 and attempt < retry_count - 1:
self._token = None
headers['Authorization'] = f'Bearer {self._get_token()}'
continue
# Handle rate limiting with exponential backoff
if response.status_code == 429:
wait_time = (2 ** attempt) + 0.5
wait_time = min(wait_time, 16)
time.sleep(wait_time)
continue
# Handle server errors
if response.status_code >= 500 and attempt < retry_count - 1:
wait_time = (2 ** attempt) + 0.5
time.sleep(wait_time)
continue
break
return response
# Health Check
def ping(self) -> Dict[str, Any]:
"""Check API health."""
response = self._request('GET', '/ping')
response.raise_for_status()
return response.json()
# User Management
def register_user(
self,
email: str,
firstName: str,
lastName: str,
organizationId: Optional[int] = None
) -> Dict[str, Any]:
"""Register a new user."""
data = {
'email': email,
'firstName': firstName,
'lastName': lastName
}
if organizationId:
data['organizationId'] = organizationId
response = self._request('POST', '/users/register', json_data=data)
response.raise_for_status()
return response.json()
def get_user(self, uuid: str) -> Dict[str, Any]:
"""Get user profile."""
response = self._request('GET', f'/users/{uuid}', user_uuid=uuid)
response.raise_for_status()
return response.json()
def update_user(
self,
uuid: str,
**fields
) -> Dict[str, Any]:
"""Update user profile. Pass any updatable fields as keyword arguments."""
response = self._request(
'PATCH', f'/users/{uuid}',
user_uuid=uuid,
json_data=fields
)
response.raise_for_status()
return response.json()
# Organizations
def get_organizations(
self,
user_uuid: str,
page: int = 1,
items_per_page: int = 30
) -> Dict[str, Any]:
"""List organizations."""
response = self._request(
'GET', '/organizations',
user_uuid=user_uuid,
params={'page': page, 'itemsPerPage': items_per_page}
)
response.raise_for_status()
return response.json()
# Course Catalog
def get_categories(self, user_uuid: str) -> List[Dict[str, Any]]:
"""List course categories."""
response = self._request('GET', '/categories', user_uuid=user_uuid)
response.raise_for_status()
return response.json()['hydra:member']
def get_courses(
self,
user_uuid: str,
category: Optional[int] = None,
page: int = 1,
items_per_page: int = 30
) -> Dict[str, Any]:
"""List courses."""
params = {'page': page, 'itemsPerPage': items_per_page}
if category:
params['category'] = category
response = self._request(
'GET', '/courses',
user_uuid=user_uuid,
params=params
)
response.raise_for_status()
return response.json()
def get_course(self, user_uuid: str, course_id: int) -> Dict[str, Any]:
"""Get course details."""
response = self._request(
'GET', f'/courses/{course_id}',
user_uuid=user_uuid
)
response.raise_for_status()
return response.json()
# Modules
def get_modules(
self,
user_uuid: str,
course: Optional[int] = None,
page: int = 1,
items_per_page: int = 30
) -> Dict[str, Any]:
"""List modules."""
params = {'page': page, 'itemsPerPage': items_per_page}
if course:
params['course'] = course
response = self._request(
'GET', '/modules',
user_uuid=user_uuid,
params=params
)
response.raise_for_status()
return response.json()
def get_module(self, user_uuid: str, module_id: int) -> Dict[str, Any]:
"""Get module details."""
response = self._request(
'GET', f'/modules/{module_id}',
user_uuid=user_uuid
)
response.raise_for_status()
return response.json()
def get_module_content(
self,
user_uuid: str,
module_id: int
) -> Dict[str, Any]:
"""Access module content."""
response = self._request(
'GET', f'/modules/{module_id}/content',
user_uuid=user_uuid
)
response.raise_for_status()
return response.json()
def refresh_module_url(
self,
user_uuid: str,
module_id: int
) -> Dict[str, Any]:
"""Re-sign an expiring content URL without resetting progress (v3.3).
Poll only when expires_at is near (e.g. < 60s away), not on a
fixed interval.
"""
response = self._request(
'GET', f'/modules/{module_id}/refresh-url',
user_uuid=user_uuid
)
response.raise_for_status()
return response.json()
def complete_module(self, user_uuid: str, module_id: int) -> None:
"""Signal learner completion for video/document/resource modules (v3.4).
FSRI emits the xAPI `completed` statement server-side. Idempotent —
repeat calls on a completed module are no-ops. Raises on 409 for
SCORM modules, which emit their own completion.
"""
response = self._request(
'POST', f'/modules/{module_id}/complete',
user_uuid=user_uuid
)
response.raise_for_status() # expect 202 Accepted
# Progress
def get_progress(
self,
user_uuid: str,
course: Optional[int] = None,
status: Optional[str] = None,
page: int = 1,
items_per_page: int = 30
) -> Dict[str, Any]:
"""Get course progress."""
params = {'page': page, 'itemsPerPage': items_per_page}
if course:
params['course'] = course
if status:
params['status'] = status
response = self._request(
'GET', '/course-progresses',
user_uuid=user_uuid,
params=params
)
response.raise_for_status()
return response.json()
# Transcripts
def get_transcripts(
self,
user_uuid: str,
page: int = 1,
items_per_page: int = 30
) -> Dict[str, Any]:
"""Get user transcripts."""
response = self._request(
'GET', '/transcripts',
user_uuid=user_uuid,
params={'page': page, 'itemsPerPage': items_per_page}
)
response.raise_for_status()
return response.json()
def download_transcript_pdf(
self,
user_uuid: str,
output_path: str
) -> None:
"""Download full transcript PDF."""
response = self._request(
'GET', '/transcript/pdf',
user_uuid=user_uuid
)
response.raise_for_status()
with open(output_path, 'wb') as f:
f.write(response.content)
def download_certificate_pdf(
self,
user_uuid: str,
transcript_id: int,
output_path: str
) -> None:
"""Download certificate PDF."""
response = self._request(
'GET', f'/transcripts/{transcript_id}/certificate/pdf',
user_uuid=user_uuid
)
response.raise_for_status()
with open(output_path, 'wb') as f:
f.write(response.content)
def deactivate_user(self, user_uuid: str) -> Dict[str, Any]:
"""Soft-deactivate a user (reversible). Returns the updated User resource."""
response = self._request(
'PATCH', f'/users/{user_uuid}',
user_uuid=user_uuid,
json_data={'status': 'inactive'}
)
response.raise_for_status()
return response.json()
def reactivate_user(self, user_uuid: str) -> Dict[str, Any]:
"""Reverse a soft-deactivation by setting status back to active."""
response = self._request(
'PATCH', f'/users/{user_uuid}',
user_uuid=user_uuid,
json_data={'status': 'active'}
)
response.raise_for_status()
return response.json()
def delete_user(self, user_uuid: str) -> None:
"""Hard-delete a user the partner registered (GDPR erasure). Irreversible."""
response = self._request(
'DELETE', f'/users/{user_uuid}',
user_uuid=user_uuid
)
response.raise_for_status() # expect 204 No Content
def sync_updated(
self,
user_uuid: str,
collection: str,
cursor: datetime,
strictly_after: bool = False
) -> List[Dict[str, Any]]:
"""Fetch all records from a filterable collection updated after ``cursor``.
Supported collections (v3.1): 'categories', 'courses', 'modules',
'course-progresses', 'transcripts'. Returns the list of changed records;
the caller is responsible for advancing the cursor.
"""
param = 'updatedAt[strictly_after]' if strictly_after else 'updatedAt[after]'
records: List[Dict[str, Any]] = []
page = 1
while True:
response = self._request(
'GET', f'/{collection}',
user_uuid=user_uuid,
params={param: cursor.isoformat().replace('+00:00', 'Z'), 'page': page}
)
response.raise_for_status()
data = response.json()
records.extend(data.get('hydra:member', []))
if 'hydra:next' not in data.get('hydra:view', {}):
break
page += 1
return records
# Example usage
if __name__ == '__main__':
import os
client = LMSClient(
client_id=os.environ['CLIENT_ID'],
client_secret=os.environ['CLIENT_SECRET'],
api_key=os.environ['API_KEY']
)
# Test connectivity
print("Testing connectivity...")
result = client.ping()
print(f"API Status: {result['status']}")
# Register a user
print("\nRegistering user...")
user = client.register_user(
email='test@example.com',
firstName='Test',
lastName='User'
)
user_uuid = user['uuid']
print(f"User UUID: {user_uuid}")
# List courses
print("\nListing courses...")
courses = client.get_courses(user_uuid)
for course in courses['hydra:member'][:3]:
print(f" - {course['title']}")
# Get progress
print("\nGetting progress...")
progress = client.get_progress(user_uuid)
print(f"Progress records: {progress['hydra:totalItems']}")
# Delta sync — only courses changed in the last 24 hours
print("\nDelta-syncing courses...")
cursor = datetime.now(timezone.utc) - timedelta(days=1)
changed = client.sync_updated(user_uuid, 'courses', cursor)
print(f"Changed courses since {cursor.isoformat()}: {len(changed)}")
Save the Client¶
Save the client class as lms_client.py and use it in your project: