Add backend interview subdirectory

backend/README.md

@@ -0,0 +1,56 @@
+# Fetch URL Challenge 
+
+The challenge: Implement a function that fetches a list of URLs concurrently with rate limiting, retries, and error handling.
+
+## Guide
+
+### Step 1: Setup
+
+This interview using a locally-running server running on `localhost:3000` that your solution code will communicate with. To start it, in a separate terminal type:
+
+```bash
+cd typescript
+npm install
+npm run server          # keep running in a terminal
+```
+
+Open another terminal window in this folder that you can use to test your solution later.
+
+For Python, open up `python/solution.py`
+
+For Typescript, open up `typescript/solution.py`
+
+### How to Run Your Solution
+
+First, make sure your server from above is running in another terminal window.
+Second, in a new terminal window, use the following commands
+
+**TypeScript**
+
+```bash
+cd typescript
+npx tsx run.ts                              # default: ./solution.ts
+npx tsx run.ts path/to/solution.ts          # or specify a file
+```
+
+**Python**
+
+```bash
+cd python
+uv run run.py                               # default: ./solution.py
+uv run run.py path/to/solution.py           # or specify a file
+```
+
+Use `run` to iterate on your solution — it prints each result so you can see
+what's happening. Use the test harness when you're ready to validate.
+
+If you need to log to stdout, the typical ways will work:
+**Typescript**
+```javascript
+console.log("foo")
+```
+
+**Python**
+```python
+print("bar")
+```

backend/python/pyproject.toml

@@ -0,0 +1,7 @@
+[project]
+name = "concurrent-fetch-challenge"
+version = "0.1.0"
+requires-python = ">=3.11,<3.14"
+dependencies = [
+    "aiohttp>=3.9",
+]

backend/python/run.py

@@ -0,0 +1,88 @@
+"""
+Runner — calls your fetch_all with the server's URL list and prints results.
+Use this to experiment before running the test harness.
+
+Usage:  uv run run.py [path/to/solution.py]
+"""
+
+import asyncio
+import importlib.util
+import json
+import sys
+import time
+import os
+import urllib.request
+from pathlib import Path
+
+solution_path = Path(sys.argv[1] if len(sys.argv) > 1 else "solution.py").resolve()
+spec = importlib.util.spec_from_file_location("solution", solution_path)
+assert spec and spec.loader, f"Could not load {solution_path}"
+_mod = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(_mod)
+fetch_all = _mod.fetch_all  # type: ignore[attr-defined]
+Success = _mod.Success  # type: ignore[attr-defined]
+Failure = _mod.Failure  # type: ignore[attr-defined]
+
+SERVER = os.environ.get("SERVER_URL", "http://localhost:3000")
+
+
+def _get(path: str):
+    with urllib.request.urlopen(f"{SERVER}{path}") as resp:
+        return json.loads(resp.read())
+
+
+def _post(path: str):
+    req = urllib.request.Request(f"{SERVER}{path}", method="POST")
+    with urllib.request.urlopen(req) as resp:
+        return json.loads(resp.read())
+
+
+async def main() -> None:
+    try:
+        urls: list[str] = _get("/urls")
+    except Exception as exc:
+        print(f"Could not reach server at {SERVER}: {exc}")
+        sys.exit(1)
+
+    _post("/reset")
+
+    print(f"Solution: {solution_path}")
+    print(f"Fetching {len(urls)} URLs...\n")
+
+    start = time.perf_counter()
+    try:
+        results = await fetch_all(
+            urls,
+            max_concurrent=10,
+            max_requests_per_second=15,
+        )
+
+        elapsed = time.perf_counter() - start
+
+        if not isinstance(results, list):
+            print(f"fetch_all returned {type(results).__name__}, expected list")
+            sys.exit(1)
+
+        print(f"Got {len(results)} results in {elapsed:.2f}s\n")
+
+        for i, r in enumerate(results):
+            if r is None:
+                print(f"  [{i}] None")
+            elif isinstance(r, Failure):
+                print(f"  [{i}] FAIL   {r.url} — {r.error}")
+            elif isinstance(r, Success):
+                print(f"  [{i}] OK     {r.url}")
+            else:
+                print(f"  [{i}] ???    {r!r}")
+
+        stats = _get("/stats")
+        print(f"\nServer stats: {json.dumps(stats, indent=2)}")
+
+    except Exception as exc:
+        elapsed = time.perf_counter() - start
+        print(f"\nfetch_all raised after {elapsed:.2f}s: {exc}")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

backend/python/solution.py

@@ -0,0 +1,49 @@
+"""
+Concurrent Fetch with Rate Limiting
+====================================
+
+Implement ``fetch_all`` to fetch a list of URLs concurrently with these
+constraints:
+
+  - Maximum ``max_concurrent`` requests in-flight at once.
+  - Maximum ``max_requests_per_second`` requests *initiated* per second.
+  - Retry transient failures (HTTP 5xx) up to 3 times.
+    Do NOT retry client errors (4xx).
+  - Return a ``Result`` for every input URL — the function must never raise.
+  - Preserve input ordering: ``results[i]`` corresponds to ``urls[i]``.
+
+You may use ``aiohttp`` and the Python standard library.
+"""
+
+from dataclasses import dataclass
+from typing import Union
+
+
+@dataclass
+class Success:
+    """A successful HTTP 200 response."""
+
+    url: str
+    body: str
+
+
+@dataclass
+class Failure:
+    """Any non-200 outcome: HTTP error status, network error, timeout, etc."""
+
+    url: str
+    error: str
+
+
+Result = Union[Success, Failure]
+
+
+async def fetch_all(
+    urls: list[str],
+    *,
+    max_concurrent: int,
+    max_requests_per_second: int,
+) -> list[Result]:
+    """Fetch all URLs concurrently and return a Result for each, preserving order."""
+    # TODO: implement
+    raise NotImplementedError