Implement JSON parsing and HTTP integration tasks
StripeSoftware Engineer · Technical ScreenLast updated
Tests practical software engineering skills in parsing JSON data, constructing batch HTTP requests, mapping returned attributes (such as color codes to coordinate points), and handling API errors.
Problem Statement
You are building a service for a geospatial analytics dashboard. You need to read point coordinates from a JSON-formatted string, validate the entries, query an external color-assignment API in batches, and output a mapping of each valid point ID to its assigned hex color.
Task
Implement the class PointColorService with the following method:
class PointColorService:
def __init__(self, http_client: HttpClient, batch_size: int = 10):
"""
http_client: An object providing a post(url: str, json_data: dict) -> HttpResponse method.
batch_size: Maximum number of points to send in a single HTTP request.
"""
pass
def assign_colors(self, json_payload: str) -> dict[str, str]:
"""
Parses points from json_payload, fetches color codes via http_client,
and returns a dict mapping point_id -> hex_color.
"""
pass
Specifications
-
Input JSON Format: The input string is expected to parse into a dictionary containing a
"points"list:{ "points": [ {"id": "pt_1", "x": 12.5, "y": -4.2}, {"id": "pt_2", "x": 0.0, "y": 15.1} ] }- Points with missing
id,x, ory, non-stringid, or non-numericx/yvalues are invalid and must be skipped. - If the JSON itself is malformed or not a valid object with a list under
"points", return an empty dictionary{}.
- Points with missing
-
HTTP Integration:
- For every batch of up to
batch_sizevalid points, make aPOSTrequest to"/api/v1/colors/resolve"usinghttp_client.post(url, json_data). - Payload structure:
{"items": [{"id": "...", "x": ..., "y": ...}, ...]}. - The response object
HttpResponsehas fields:status_code: intandbody: dict. - On success (
status_code == 200),bodyis structured as{"colors": [{"id": "...", "color": "#RRGGBB"}, ...]}. - If an HTTP call fails (
status_code != 200or an exception occurs), assign a fallback color"#000000"to all points in that batch.
- For every batch of up to
-
Output Format:
- Return a dictionary mapping
point_idto its resolved hex color string (e.g.{"pt_1": "#FF5733", "pt_2": "#33FF57"}).
- Return a dictionary mapping
Constraints
0 <= number of points <= 10,0001 <= batch_size <= 500- Point IDs within a valid payload are unique.
Example
# Input payload
payload = '''{
"points": [
{"id": "p1", "x": 1.0, "y": 2.0},
{"id": "p2", "x": 3.0, "y": "invalid"},
{"id": "p3", "x": -1.0, "y": 0.0}
]
}'''
# Assuming batch_size=2 and http_client returns:
# POST /api/v1/colors/resolve with items [p1, p3] -> status 200,
# body: {"colors": [{"id": "p1", "color": "#FF0000"}, {"id": "p3", "color": "#0000FF"}]}
service = PointColorService(http_client, batch_size=2)
result = service.assign_colors(payload)
# result == {"p1": "#FF0000", "p3": "#0000FF"}
Hints
Hint 1
Start by parsing the raw JSON string safely inside a try-except block, validating each point's structure before accumulating valid items.
Hint 2
Slice your valid points list into chunks of length batch_size to construct each HTTP request payload.
Hint 3
Safely wrap the HTTP client call in a try-except block, checking for both response status codes and network exceptions to fall back to '#000000'.
Follow-up questions
- How would you handle rate-limiting (e.g. HTTP 429 with a 'Retry-After' header) and transient 5xx errors using exponential backoff?
- If the input data size exceeds memory (e.g., several gigabytes of stream data), how would you refactor the JSON parsing and request pipeline?
- How would you run batch HTTP requests concurrently while respecting a client-side concurrency cap?
Solution walkthrough is for members
Get the full walkthrough with reference code and complexity analysis, plus Kitty Coach at 300 messages a day. Hints, the code runner and community answers stay free.