Learnosity · walsh-conor · Oct 21, 2025 · Oct 21, 2025 · Oct 21, 2025 · Oct 21, 2025
@@ -119,6 +119,7 @@ Following are the routes to access our APIs.
 * Items API : http://localhost:8000/itemsapi
 * Reports API : http://localhost:8000/reportsapi
 * Question Editor API : http://localhost:8000/questioneditorapi
+* Data API : http://localhost:8000/dataapi
 
 Open these pages with your web browser. These are all basic examples of Learnosity's integration. You can interact with these demo pages to try out the various APIs. The Items API example is a basic example of an assessment loaded into a web page with Learnosity's assessment player. You can interact with this demo assessment to try out the various Question types.
 

@@ -5,12 +5,13 @@
 # with `rendering_type: "assess"`.
 
 # Include server side Learnosity SDK, and set up variables related to user access
-from learnosity_sdk.request import Init
+from learnosity_sdk.request import Init, DataApi
 from learnosity_sdk.utils import Uuid
 from .. import config # Load consumer key and secret from config.py
 # Include web server and Jinja templating libraries.
 from http.server import BaseHTTPRequestHandler, HTTPServer
 from jinja2 import Template
+import json
 
 # - - - - - - Section 1: Learnosity server-side configuration - - - - - - #
 
@@ -375,6 +376,10 @@ def do_GET(self) -> None:
                             <td>Author Aide API</td>
                             <td><a href="/authoraideapi">Here</a></td>
                         </tr>
+                        <tr>
+                            <td>Data API</td>
+                            <td><a href="/dataapi">Here</a></td>
+                        </tr>
                     </table>
                 </body>
             </html>
@@ -542,6 +547,175 @@ def do_GET(self) -> None:
 
                     self.createResponse(response)
 
+        if self.path.endswith("/dataapi"):
+            # Data API demo - retrieve items from the itembank
+            template = Template("""<!DOCTYPE html>
+            <html>
+                <head>
+                    <style>
+                        body { font-family: Arial, sans-serif; margin: 20px; }
+                        h1 { color: #333; }
+                        .demo-section {
+                            margin: 30px 0;
+                            padding: 20px;
+                            border: 1px solid #ddd;
+                            border-radius: 5px;
+                            background-color: #f9f9f9;
+                        }
+                        .demo-section h2 {
+                            color: #0066cc;
+                            margin-top: 0;
+                        }
+                        .item {
+                            margin: 10px 0;
+                            padding: 10px;
+                            background-color: white;
+                            border-left: 3px solid #0066cc;
+                        }
+                        .item-reference {
+                            font-weight: bold;
+                            color: #0066cc;
+                        }
+                        .item-status {
+                            color: #666;
+                            font-size: 0.9em;
+                        }
+                        .meta-info {
+                            background-color: #e8f4f8;
+                            padding: 10px;
+                            margin: 10px 0;
+                            border-radius: 3px;
+                            font-family: monospace;
+                            font-size: 0.9em;
+                        }
+                        .error {
+                            color: #cc0000;
+                            background-color: #ffe6e6;
+                            padding: 10px;
+                            border-radius: 3px;
+                        }
+                        pre {
+                            background-color: #f5f5f5;
+                            padding: 10px;
+                            border-radius: 3px;
+                            overflow-x: auto;
+                        }
+                    </style>
+                </head>
+                <body>
+                    <h1>{{ name }}</h1>
+                    <p>This demo shows how to use the Data API to retrieve items from the Learnosity itembank.</p>
+
+                    <div class="demo-section">
+                        <h2>Demo 1: Manual Iteration (5 items)</h2>
+                        <p>Using <code>request()</code> method with manual pagination via the 'next' pointer.</p>
+                        {{ demo1_output }}
+                    </div>
+
+                    <div class="demo-section">
+                        <h2>Demo 2: Page Iteration (5 pages)</h2>
+                        <p>Using <code>request_iter()</code> method to automatically iterate over pages.</p>
+                        {{ demo2_output }}
+                    </div>
+
+                    <div class="demo-section">
+                        <h2>Demo 3: Results Iteration (5 items)</h2>
+                        <p>Using <code>results_iter()</code> method to automatically iterate over individual items.</p>
+                        {{ demo3_output }}
+                    </div>
+
+                    <p><a href="/">Back to API Examples</a></p>
+                </body>
+            </html>
+            """)
+
+            # Run the Data API demos
+            itembank_uri = 'https://data.learnosity.com/latest-lts/itembank/items'
+            security_packet = {
+                'consumer_key': config.consumer_key,
+                'domain': host,
+            }
+            data_api = DataApi()
+
+            # Demo 1: Manual iteration
+            demo1_html = ""
+            try:
+                data_request = {'limit': 1}
+                for i in range(5):
+                    result = data_api.request(itembank_uri, security_packet,
+                                             config.consumer_secret, data_request, 'get')
+                    response = result.json()
+
+                    if response.get('data'):
+                        item = response['data'][0]
+                        demo1_html += f"""
+                        <div class="item">
+                            <div class="item-reference">Item {i+1}: {item.get('reference', 'N/A')}</div>
+                            <div class="item-status">Status: {item.get('status', 'N/A')}</div>
+                        </div>
+                        """
+
+                    if response.get('meta', {}).get('next'):
+                        data_request['next'] = response['meta']['next']
+                    else:
+                        break
+            except Exception as e:
+                demo1_html = f'<div class="error">Error: {str(e)}</div>'
+
+            # Demo 2: Page iteration
+            demo2_html = ""
+            try:
+                data_request = {'limit': 1}
+                page_count = 0
+                for page in data_api.request_iter(itembank_uri, security_packet,
+                                                  config.consumer_secret, data_request, 'get'):
+                    page_count += 1
+                    demo2_html += f"""
+                    <div class="meta-info">
+                        Page {page_count}: {len(page.get('data', []))} items
+                    </div>
+                    """
+                    if page.get('data'):
+                        for item in page['data']:
+                            demo2_html += f"""
+                            <div class="item">
+                                <div class="item-reference">{item.get('reference', 'N/A')}</div>
+                                <div class="item-status">Status: {item.get('status', 'N/A')}</div>
+                            </div>
+                            """
+                    if page_count >= 5:
+                        break
+            except Exception as e:
+                demo2_html = f'<div class="error">Error: {str(e)}</div>'
+
+            # Demo 3: Results iteration
+            demo3_html = ""
+            try:
+                data_request = {'limit': 1}
+                result_count = 0
+                for item in data_api.results_iter(itembank_uri, security_packet,
+                                                  config.consumer_secret, data_request, 'get'):
+                    result_count += 1
+                    demo3_html += f"""
+                    <div class="item">
+                        <div class="item-reference">Item {result_count}: {item.get('reference', 'N/A')}</div>
+                        <div class="item-status">Status: {item.get('status', 'N/A')}</div>
+                        <pre>{json.dumps(item, indent=2)[:500]}...</pre>
+                    </div>
+                    """
+                    if result_count >= 5:
+                        break
+            except Exception as e:
+                demo3_html = f'<div class="error">Error: {str(e)}</div>'
+
+            response = template.render(
+                name='Data API Example',
+                demo1_output=demo1_html,
+                demo2_output=demo2_html,
+                demo3_output=demo3_html
+            )
+            self.createResponse(response)
+
 def main() -> None:
     web_server = HTTPServer((host, port), LearnosityServer)
     print("Server started http://%s:%s. Press Ctrl-c to quit." % (host, port))

@@ -2,13 +2,55 @@
 from requests import Response
 import requests
 import copy
+import re
+from urllib.parse import urlparse
 
 from learnosity_sdk.exceptions import DataApiException
 from learnosity_sdk.request import Init
 
 
 class DataApi(object):
 
+    def _extract_consumer(self, security_packet: Dict[str, str]) -> str:
+        """
+        Extract the consumer key from the security packet.
+
+        Args:
+            security_packet (dict): The security object
+
+        Returns:
+            string: The consumer key
+        """
+        return security_packet.get('consumer_key', '')
+
+    def _derive_action(self, endpoint: str, action: str) -> str:
+        """
+        Derive the action metadata from the endpoint and action parameter.
+
+        The action format is: {action}_{endpoint_path}
+        For example: 'get_/itembank/items' or 'set_/itembank/activities'
+
+        Args:
+            endpoint (string): The full url to the endpoint
+            action (string): 'get', 'set', 'update', etc.
+
+        Returns:
+            string: The derived action string
+        """
+        # Parse the URL to extract the path
+        parsed_url = urlparse(endpoint)
+        path = parsed_url.path.rstrip("/")
+
+        # Remove version prefix if present (e.g., /v1, /v2, /latest)
+        # Only match 'v' followed by digits (v1, v2, etc.) or 'latest'
+        path_parts = path.split('/')
+        if len(path_parts) > 1:
+            first_segment = path_parts[1].lower()
+            if re.fullmatch(r"v\d+", first_segment) or first_segment == "latest":
+                path = '/' + '/'.join(path_parts[2:])
+
+        return f"{action}_{path}"
+
     def request(self, endpoint: str, security_packet: Dict[str, str],
                 secret: str, request_packet:Dict[str, Any] = {}, action: str = 'get') -> Response:
         """
@@ -33,7 +75,18 @@ def request(self, endpoint: str, security_packet: Dict[str, str],
             see http://docs.python-requests.org/en/master/api/#requests.Request
         """
         init = Init('data', security_packet, secret, request_packet, action)
-        return requests.post(endpoint, data=init.generate())
+
+        # Extract metadata for routing
+        consumer = self._extract_consumer(security_packet)
+        derived_action = self._derive_action(endpoint, action)
+
+        # Add metadata as HTTP headers for ALB routing
+        headers = {
+            'X-Learnosity-Consumer': consumer,
+            'X-Learnosity-Action': derived_action
+        }
+
+        return requests.post(endpoint, data=init.generate(), headers=headers)
 
     def results_iter(self, endpoint: str, security_packet: Dict[str, str],
                      secret: str, request_packet: Dict[str, Any] = {},

@@ -58,6 +58,12 @@ def test_request(self) -> None:
         assert responses.calls[0].request.url == self.endpoint
         assert 'signature' in cast(Dict[str, Any], responses.calls[0].request.body)
 
+        # Verify metadata headers are present
+        assert 'X-Learnosity-Consumer' in responses.calls[0].request.headers
+        assert responses.calls[0].request.headers['X-Learnosity-Consumer'] == 'yis0TYCu7U9V4o7M'
+        assert 'X-Learnosity-Action' in responses.calls[0].request.headers
+        assert responses.calls[0].request.headers['X-Learnosity-Action'] == 'get_/itembank/items'
+
     @responses.activate
     def test_request_iter(self) -> None:
         """Verify that `request_iter` returns an iterator of pages"""
@@ -123,3 +129,76 @@ def test_results_iter_invalid_response_data(self) -> None:
         with self.assertRaisesRegex(DataApiException, "server returned invalid json: "):
             list(client.results_iter(self.endpoint, self.security, self.consumer_secret,
                                      self.request, self.action))
+
+    def test_extract_consumer(self) -> None:
+        """Verify that consumer key is correctly extracted from security packet"""
+        client = DataApi()
+        consumer = client._extract_consumer(self.security)
+        assert consumer == 'yis0TYCu7U9V4o7M'
+
+    def test_extract_consumer_missing(self) -> None:
+        """Verify that empty string is returned when consumer_key is missing"""
+        client = DataApi()
+        consumer = client._extract_consumer({})
+        assert consumer == ''
+
+    def test_derive_action_with_version(self) -> None:
+        """Verify that action is correctly derived from endpoint with version"""
+        client = DataApi()
+        action = client._derive_action('https://data.learnosity.com/v1/itembank/items', 'get')
+        assert action == 'get_/itembank/items'
+
+    def test_derive_action_with_latest(self) -> None:
+        """Verify that action is correctly derived from endpoint with 'latest' version"""
+        client = DataApi()
+        action = client._derive_action('https://data.learnosity.com/latest/itembank/questions', 'get')
+        assert action == 'get_/itembank/questions'
+
+    def test_derive_action_without_version(self) -> None:
+        """Verify that action is correctly derived from endpoint without version"""
+        client = DataApi()
+        action = client._derive_action('https://data.learnosity.com/itembank/activities', 'set')
+        assert action == 'set_/itembank/activities'
+
+    def test_derive_action_session_scores(self) -> None:
+        """Verify that action is correctly derived for session_scores endpoint"""
+        client = DataApi()
+        action = client._derive_action('https://data.learnosity.com/v1/session_scores', 'get')
+        assert action == 'get_/session_scores'
+
+    def test_derive_action_route_starting_with_v(self) -> None:
+        """Verify that routes starting with 'v' but not version numbers are not stripped"""
+        client = DataApi()
+        # Routes like /valid, /vendors, /version should NOT be treated as version prefixes
+        action = client._derive_action('https://data.learnosity.com/valid/route', 'get')
+        assert action == 'get_/valid/route'
+
+    def test_derive_action_with_v2(self) -> None:
+        """Verify that v2 version prefix is correctly stripped"""
+        client = DataApi()
+        action = client._derive_action('https://data.learnosity.com/v2/itembank/items', 'get')
+        assert action == 'get_/itembank/items'
+
+    def test_derive_action_with_v_only(self) -> None:
+        """Verify that a route segment of just 'v' is not treated as a version"""
+        client = DataApi()
+        action = client._derive_action('https://data.learnosity.com/v/items', 'get')
+        assert action == 'get_/v/items'
+
+    @responses.activate
+    def test_metadata_headers_in_paginated_requests(self) -> None:
+        """Verify that metadata headers are sent in all paginated requests"""
+        for dummy in self.dummy_responses:
+            responses.add(responses.POST, self.endpoint, json=dummy)
+        client = DataApi()
+        # Consume the iterator to trigger the requests
+        list(client.request_iter(self.endpoint, self.security, self.consumer_secret,
+                                 self.request, self.action))
+
+        # Verify both requests have the metadata headers
+        assert len(responses.calls) == 2
+        for call in responses.calls:
+            assert 'X-Learnosity-Consumer' in call.request.headers
+            assert call.request.headers['X-Learnosity-Consumer'] == 'yis0TYCu7U9V4o7M'
+            assert 'X-Learnosity-Action' in call.request.headers
+            assert call.request.headers['X-Learnosity-Action'] == 'get_/itembank/items'