REST APIs & Authentication.

The first three Applied Python lessons established a local environment, a configurable command-line tool and a validated file pipeline. This lesson adds a new boundary: remote services can be slow, reject permissions, send only part of a dataset or return an unexpected response. You'll learn to make those conditions visible instead of publishing an inaccurate report.

• Explain HTTP request methods, resource URLs, headers, status codes and JSON response contracts.
• Distinguish 200 from 204 and authentication failures from permission failures.
• Construct properly encoded URL query strings and enforce an approved HTTPS origin for real remote clients.
• Describe API keys, OAuth access tokens, least-privilege delegated versus application authorization, and secret redaction.
• Separate transport timeouts, HTTP errors, JSON decode errors and malformed response schemas.
• Collect bounded, validated paginated records while rejecting repeated or unapproved next links.
• Interpret 429 throttling and Retry-After guidance using safe, bounded retry policies for read-only requests.
• Connect a fake response provider or optional localhost HTTP mock to deterministic offline inventory reporting.

Prerequisites: Complete Filesystem & Structured Data Pipelines and Command-Line Tools & Configuration, or be comfortable with functions, dictionaries, JSON, explicit argument lists and error handling. This lesson is about writing trustworthy read-only clients, not acquiring permissions for a real production API.

Your CLI can now read and validate local files. Remote systems expose APIs so a program can request information instead of opening another application's data files. An HTTP request specifies a method, URL, headers and sometimes a body. A response supplies a status, headers and optional response body. REST APIs commonly represent resources using URLs and JSON, but each provider defines its own contract. A successful connection is not evidence that you requested the right resource or received the expected data.

from urllib.parse import urlsplit
url='https://api.example.invalid/v1/devices?site=HQ'
parts=urlsplit(url)
print(parts.scheme, parts.path, parts.query)

HTTP response status changes what your client should do next. A 200 OK response may contain JSON, whereas 204 No Content indicates a successful response without a body, so calling json.loads('') is incorrect. Authentication failures (401) differ from insufficient permission (403). A 404 may indicate a missing resource or an endpoint you cannot see; 429 signals throttling and a 5xx response means the server couldn't complete the request. Don't classify every non-200 response as a network outage, and don't classify every 200 response as a valid application result.

def status_kind(code):
    if code == 204: return 'no-content'
    if code == 401: return 'authentication'
    if code == 403: return 'permission'
    if code == 429: return 'throttled'
    return 'success' if 200 <= code < 300 else 'error'
print(status_kind(401))

{
  "method": "GET",
  "url": "https://api.example.invalid/v1/devices?site=HQ",
  "headers": {
    "Accept": "application/json",
    "Authorization": "Bearer <redacted fictional placeholder>"
  },
  "timeoutSeconds": 5
}

Build URLs from a trusted base and encode query parameter values using urllib.parse.urlencode rather than concatenating untrusted strings. Spaces, ampersands and other characters have special roles in URL syntax. A site named 'HQ & East' should remain one query value, not become another parameter. Keep an explicit HTTPS origin allowlist for operational clients and validate hostnames before following URLs or redirects, especially pagination URLs returned by a server. Query strings often appear in proxies, histories and logs; never put access tokens or passwords there.

from urllib.parse import urlencode
from urllib.request import Request
url='https://api.example.invalid/v1/devices?'+urlencode({'site':'HQ & East'})
request=Request(url,headers={'Accept':'application/json'},method='GET')
print(request.full_url.endswith('site=HQ+%26+East'))

from urllib.parse import urlsplit

parts = urlsplit('https://api.example.invalid:443/v1/devices?page=2')
print(parts.scheme)
print(parts.hostname)
print(parts.port)
print(parts.username is None and parts.password is None)
print(parts.path)

Authentication establishes an identity for a request; authorization determines what that identity can access. An API key identifies a caller only within its provider's defined scheme and may not represent a human user. OAuth 2.0 access tokens typically carry delegated or application permissions and are presented as Bearer tokens, but bearer possession alone grants access only within the token's audience, lifetime and scope. Microsoft Graph scenarios distinguish delegated user context from application-only permissions; an API call should request the minimum approved permissions for its work. Credentials are not necessary for our fictional offline simulator.

def classify_auth(status):
    return {401:'Check authentication',403:'Review permissions'}.get(status,'Other status')
print(classify_auth(403))

A bearer token is sensitive because anyone who obtains a usable one may be able to act with its permissions. Never embed real tokens in source control, lesson examples, browser code or query strings. Local environment variables may be practical for a short demonstration, but they are not a secret vault and may leak through process inspection or CI logs; production tools should use organization-approved credential management, rotating short-lived tokens and approved libraries or SDKs. Build an Authorization header only from a provided token value, avoid printing it, and redact known sensitive fields from errors.

def safe_headers(headers):
    return {k:('<redacted>' if k.lower() in {'authorization','x-api-key'} else v)
            for k,v in headers.items()}
print(safe_headers({'Authorization':'Bearer DEMO','Accept':'application/json'}))

Even a correct API endpoint can be unreachable, slow or interrupted. `urllib.request.urlopen` accepts a timeout in seconds for blocking operations; failing to provide bounded network timeouts can stall scheduled jobs. Distinguish socket.timeout/TimeoutError and URLError transport failures from HTTPError responses, which carry a status code and often a body. A 200 status paired with invalid JSON is a response validation error, not a network timeout. Keep error messages clear enough to diagnose the stage without exposing secrets or response payloads unnecessarily.

def describe_result(result):
    if result['kind']=='timeout': return 'Transport timed out'
    if result['kind']=='http': return f"HTTP {result['status']}"
    return 'Invalid response envelope'
print(describe_result({'kind':'timeout'}))

Many directory and IT inventory APIs return only a subset of records per response. Microsoft Graph uses `value` arrays and may provide `@odata.nextLink` when another page exists. A client that reads one 200 response and stops may quietly omit most devices. Validate each page's shape, collect records in a deterministic order, and enforce a maximum number of pages and records. Maintain a set of visited URLs to catch repeated/cyclic links. Validate every nextLink against the approved HTTPS service origin before fetching it; never blindly follow a server-supplied URL to an arbitrary host.

def collect(page1,page2):
    return sorted(page1['value']+page2['value'])
print(collect({'value':['SW-02']},{'value':['AP-01']}))

from urllib.parse import urlsplit

def allowed_page(url):
    try:
        p = urlsplit(url)
        return (p.scheme == 'https' and
                p.hostname == 'api.example.invalid' and
                p.port in (None, 443) and
                p.username is None and p.password is None and
                p.path == '/v1/devices')
    except ValueError:
        return False

print(allowed_page('https://api.example.invalid/v1/devices?page=2'))
print(allowed_page('https://evil.example/steal'))

APIs protect their capacity by throttling clients. Microsoft Graph commonly responds with HTTP 429 and a Retry-After header containing a suggested wait interval. For a read-only GET, a limited retry with the server's guidance (and a fallback backoff policy when guidance is absent) can be reasonable. Never immediately hammer the endpoint or sleep unboundedly. Parse Retry-After carefully: HTTP permits both delta-seconds and HTTP-date, so the numeric-only helper in this lesson is an explicit simplified training subset. Changing operations such as POST must not be automatically retried without understanding idempotency keys, duplicate creation and the API's contract.

def retry_seconds(headers):
    value = headers.get('Retry-After','')
    return min(int(value),30) if value.isdecimal() else None
print(retry_seconds({'Retry-After':'10'}))

A practical API-based inventory reporter should use your earlier CLI for approved configuration, request data from a known service with the minimum permissions, interpret HTTP errors, page through complete results, validate JSON records, then build the stable report learned in Lesson 03. Design the core as an offline-testable function that takes an injected response provider, and keep actual network access in a separate adapter. The local example uses a tiny server bound to your own loopback address with fictional responses—no Microsoft tenant, API token, customer data or third-party package needed. After verifying that behavior, learners can adapt it to an authorized sandbox API in later training.

def build_inventory_report(items):
    hosts=sorted(item['host'] for item in items)
    return {'total':len(hosts),'hosts':hosts}
print(build_inventory_report([{'host':'SW-02'},{'host':'AP-01'}]))

You can practice actual HTTP request and response behavior without a Microsoft tenant, external test account or real credentials. Save the Python script below as local_api_demo.py inside the virtual environment project from Lesson 01. It starts a temporary server on your computer's 127.0.0.1 interface with a random available port, makes GET requests using urllib.request, validates paged JSON, prints a report and shuts the server down. All devices are fictional and no files are modified.

"""Offline REST API practice. Binds ONLY to 127.0.0.1; uses fictional inventory."""
import argparse
import json
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.error import HTTPError, URLError
from urllib.parse import parse_qs, urlsplit
from urllib.request import Request, urlopen

DEVICES = [
    {"host": "SW-01", "status": "online"},
    {"host": "AP-02", "status": "offline"},
    {"host": "FW-01", "status": "online"},
]
SCENARIOS = ("normal", "auth", "forbidden", "throttle", "service", "bad-json")

class DemoAPI(BaseHTTPRequestHandler):
    def log_message(self, format, *args):
        pass  # Don't print request headers or credentials.

    def do_GET(self):
        parts = urlsplit(self.path)
        query = parse_qs(parts.query)
        scenario = query.get("scenario", ["normal"])[0]
        page = query.get("page", ["1"])[0]

        if parts.path != "/v1/devices":
            return self.send_json(404, {"error": "not-found"})
        if scenario == "auth":
            return self.send_json(401, {"error": "demo-auth-required"})
        if scenario == "forbidden":
            return self.send_json(403, {"error": "demo-permission-denied"})
        if scenario == "throttle":
            return self.send_json(429, {"error": "demo-throttled"}, {"Retry-After": "3"})
        if scenario == "service":
            return self.send_json(503, {"error": "demo-service-unavailable"})
        if scenario == "bad-json":
            return self.send_body(200, b"{not-json", "application/json")

        if page == "1":
            base = f"http://127.0.0.1:{self.server.server_port}"
            payload = {"value": DEVICES[:2], "@odata.nextLink": base + "/v1/devices?page=2"}
        elif page == "2":
            payload = {"value": DEVICES[2:]}
        else:
            return self.send_json(404, {"error": "page-not-found"})
        self.send_json(200, payload)

    def send_json(self, status, payload, extra=None):
        body = json.dumps(payload, sort_keys=True).encode("utf-8")
        self.send_body(status, body, "application/json", extra)

    def send_body(self, status, body, content_type, extra=None):
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        for key, value in (extra or {}).items():
            self.send_header(key, value)
        self.end_headers()
        self.wfile.write(body)


def read_page(url, expected_port):
    parts = urlsplit(url)
    # HTTP is intentional ONLY for the local, fake server with no secrets.
    if (parts.scheme != "http" or parts.hostname != "127.0.0.1"
            or parts.port != expected_port or parts.username or parts.password
            or parts.path != "/v1/devices"):
        raise ValueError("Blocked unexpected continuation URL")
    req = Request(url, headers={"Accept": "application/json"}, method="GET")
    try:
        with urlopen(req, timeout=4) as response:
            status = response.status
            body = response.read(64 * 1024 + 1)
            if len(body) > 64 * 1024:
                raise ValueError("Response exceeded 64 KiB limit")
    except HTTPError as exc:
        hint = exc.headers.get("Retry-After")
        raise ValueError(f"HTTP {exc.code}" + (f"; Retry-After={hint}" if hint else "")) from None
    except (URLError, TimeoutError) as exc:
        raise ValueError("Network request failed or timed out") from exc
    if status != 200:
        raise ValueError(f"Unexpected HTTP {status}")
    try:
        data = json.loads(body.decode("utf-8"))
    except (UnicodeError, json.JSONDecodeError) as exc:
        raise ValueError("Invalid JSON response") from exc
    if not isinstance(data, dict) or not isinstance(data.get("value"), list):
        raise ValueError("Invalid API response shape")
    return data


def collect(base_url, port):
    seen = set()
    hosts = []
    url = base_url
    for _ in range(5):  # hard maximum, including the last page
        if url in seen:
            raise ValueError("Pagination loop")
        seen.add(url)
        page = read_page(url, port)
        for item in page["value"]:
            if (not isinstance(item, dict)
                    or not isinstance(item.get("host"), str)
                    or not item["host"].strip()
                    or item.get("status") not in ("online", "offline")):
                raise ValueError("Invalid device record")
            hosts.append({"host": item["host"].strip(), "status": item["status"]})
        next_url = page.get("@odata.nextLink")
        if not next_url:
            return {"total": len(hosts),
                    "offline_hosts": sorted(x["host"] for x in hosts if x["status"] == "offline"),
                    "pages": len(seen)}
        if not isinstance(next_url, str):
            raise ValueError("Invalid pagination link")
        url = next_url
    raise ValueError("Page budget exceeded; no partial report allowed")


def main(argv=None):
    parser = argparse.ArgumentParser(description="Fictional API over local loopback only")
    parser.add_argument("--scenario", choices=SCENARIOS, default="normal")
    args = parser.parse_args(argv)
    server = ThreadingHTTPServer(("127.0.0.1", 0), DemoAPI)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        first = f"http://127.0.0.1:{server.server_port}/v1/devices"
        if args.scenario != "normal":
            first += "?scenario=" + args.scenario
        report = collect(first, server.server_port)
        print(json.dumps(report, sort_keys=True, indent=2))
        return 0
    except ValueError as exc:
        print("Local API demo error: " + str(exc), file=sys.stderr)
        return 1
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)


if __name__ == "__main__":
    raise SystemExit(main())

Read the read_page() function closely: it allows only the exact loopback host and port chosen by the local server. The all-or-nothing collect() loop validates every page before printing a result. A normal cloud API client will need an HTTPS origin allowlist, approved credentials, robust retry limits and auditing before being trusted with operational data.

Begin with an encoded read-only URL and work toward independent status handling, JSON schema checks, secret redaction, transport errors, rate limits, safe pagination URLs and a complete two-page inventory reporter. The workbench executes actual Python code and checks changed response fixtures without calling a live API.

The built-in Python modules and Microsoft Graph documentation underpin these examples. Each real service has its own supported endpoints, authentication requirements, paging contracts and rate limits; verify its current guidance before integrating a real environment.