P3 — Python Fundamentals II: Files, APIs, and Environment Management
Learning objective
Write Python scripts that load configuration from environment variables, call an HTTP API, parse the JSON response, and write structured output to a file — without hardcoding any credentials or URLs.
Why this matters
Every production AI service loads its API keys from environment variables, not from source code. Every API integration follows the same pattern: construct request → send HTTP call → parse response → handle errors → persist results. Understanding this pattern at the code level means you can review integration designs, spot credential-handling errors before they become incidents, and write the actual integration scripts when the engineering team needs proof-of-concept code.
Pre-work
- freeCodeCamp: Python Requests Library — search YouTube for freeCodeCamp requests tutorial, ~30 min
- python-dotenv documentation at
pypi.org/project/python-dotenv— read the README (~15 min) - Install packages in your venv:
pip install requests python-dotenv
Core concept explained
The environment variable pattern
Credentials — API keys, database passwords, service URLs — must never appear in source code. The pattern used across all professional Python services is: store credentials in a .env file locally (never committed to Git), load them at runtime using python-dotenv, and access them via os.environ. In production cloud deployments, the .env file is replaced by a secrets manager (AWS Secrets Manager, Azure Key Vault) — but the application code uses the same os.environ access pattern regardless. Building this habit in Week P3 means you never have to retrofit it.
# .env file — save this as .env in your project root (never commit this file)
ANTHROPIC_API_KEY=sk-ant-...
BASE_URL=https://api.anthropic.com# Python code — add this to the top of your script
import os
from dotenv import load_dotenv
load_dotenv() # loads .env into os.environ
api_key = os.environ.get("ANTHROPIC_API_KEY")
if not api_key:
raise EnvironmentError("ANTHROPIC_API_KEY not set — check your .env file")The requests pattern
The requests library abstracts HTTP. A POST request to the Claude API follows a fixed pattern: construct headers (authentication, content type, API version), construct the body as a Python dictionary, call requests.post(), check the response status code, and parse the JSON body. This is the same pattern used for every HTTP API regardless of vendor.
import requests
def call_claude_api(system_prompt: str, user_message: str, api_key: str) -> dict:
"""
Make a single call to the Claude messages endpoint.
Returns the full parsed JSON response or raises on HTTP error.
"""
headers = {
"x-api-key": api_key,
"anthropic-version": "2023-06-01",
"content-type": "application/json"
}
body = {
"model": "claude-haiku-4-5-20251001",
"max_tokens": 1024,
"system": system_prompt,
"messages": [{"role": "user", "content": user_message}]
}
response = requests.post(
"https://api.anthropic.com/v1/messages",
headers=headers,
json=body,
timeout=30
)
response.raise_for_status() # raises HTTPError for 4xx/5xx
return response.json()Common mistake: Using response.text and then calling json.loads() on it separately. The response.json() method handles this and raises a clear exception if the body is not valid JSON. Always use response.json().
File I/O for structured data
Reading and writing JSON files is the standard pattern for persisting AI system state, configuration, and logs. The json module handles serialisation. For append-only logs (one JSON object per line), use JSONL format — open with mode 'a' (append) and write json.dumps(record) + '\n'.
import json
from pathlib import Path
def write_json_file(data: dict, filepath: str) -> None:
"""Write a dictionary to a JSON file, creating parent directories."""
path = Path(filepath)
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
def append_jsonl(record: dict, filepath: str) -> None:
"""Append a single record to a JSONL file."""
with open(filepath, 'a', encoding='utf-8') as f:
f.write(json.dumps(record) + '\n')Step-by-step exercise — Tier 1 (Guided)
What Tier 1 means for this week: the Core Concept section contains complete, runnable implementations of every pattern used in this exercise — the load_dotenv() pattern, the requests.post() pattern, and both file helper functions. Type each one into your editor as you encounter it. Do not copy-paste from this document. The ClaudeClient class in Step 2 is larger than anything in Week P1 — type it methodically, section by section, and run the partial file through python -m py_compile after each logical block to catch typing errors early before the whole class is complete.
Deliverable: claude_client.py — a reusable Claude API client that uses prompt_utils.py (started in Week P1, extended below), loads credentials from environment variables, and persists all interactions to a JSONL log. This becomes the foundation for all Module 1–2 exercises.
Step 1 — Set up credentials
Create .env in your cca-prep/ folder (verify it's in .gitignore):
ANTHROPIC_API_KEY=your-key-here
ANTHROPIC_MODEL=claude-haiku-4-5-20251001
ANTHROPIC_API_URL=https://api.anthropic.com/v1/messagesStep 2 — Complete prompt_utils.py, then implement the client
prompt_utils.py from Week P1 has only build_message() in it so far — that was deliberate, since the four functions originally sketched out in Week P1 need patterns this week teaches. Add the remaining three now, plus append_jsonl(). Like everything else this week, these are shown complete — type them in, the same way you typed call_claude_api and the file helpers above:
def build_messages_array(system_prompt: str, user_message: str,
history: Optional[list] = None) -> dict:
"""
Build a complete API request body for the Claude messages endpoint.
Args:
system_prompt: The system instruction string
user_message: The current user input
history: Optional list of prior message dicts (role + content)
Returns:
Dictionary suitable for JSON serialisation as Claude API request body
"""
messages = list(history) if history else []
messages.append(build_message("user", user_message))
return {
"system": system_prompt,
"messages": messages
}
# Note: no "model" or "max_tokens" key here — that's not this function's
# job. The caller (ClaudeClient.complete, below) adds those to the dict
# this function returns, the same two-step split you see in Step 2.
def extract_text_response(api_response: dict) -> str:
"""
Extract the text content from a Claude API response dictionary.
Handles the nested content array structure.
Args:
api_response: The parsed JSON response from the Claude API
Returns:
The text content string, or empty string if extraction fails
"""
try:
return api_response["content"][0]["text"]
except (KeyError, IndexError, TypeError):
return ""
def save_interaction_log(system_prompt: str, user_message: str,
response_text: str, filepath: str) -> None:
"""
Append a single interaction (system/user/response) to a JSONL log file.
Each line is a valid JSON object. Creates the file if it doesn't exist.
Args:
system_prompt: The system prompt used
user_message: The user input
response_text: The extracted response text
filepath: Path to the .jsonl log file
"""
record = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"system_prompt": system_prompt,
"user_message": user_message,
"response_text": response_text
}
append_jsonl(record, filepath)
def append_jsonl(record: dict, filepath: str) -> None:
"""Append a single record to a JSONL file."""
with open(filepath, 'a', encoding='utf-8') as f:
f.write(json.dumps(record) + '\n')Add three imports to the top of prompt_utils.py, above build_message(), if they aren't there already: import json (used by append_jsonl), from typing import Optional (used by build_messages_array's history parameter), and from datetime import datetime, timezone (used by save_interaction_log's timestamp).
Add tests for the three new functions to test_prompt_utils.py, one valid and one malformed case each — same assert-and-print pattern as every previous week:
# Test build_messages_array
msgs = build_messages_array("Be helpful", "Hi there")
assert msgs["system"] == "Be helpful"
assert msgs["messages"] == [{"role": "user", "content": "Hi there"}]
print("build_messages_array no history: PASS")
msgs_with_history = build_messages_array(
"Be helpful", "And this?",
history=[{"role": "user", "content": "What is 2+2?"}, {"role": "assistant", "content": "4"}]
)
assert len(msgs_with_history["messages"]) == 3
assert msgs_with_history["messages"][-1]["content"] == "And this?"
print("build_messages_array with history: PASS")
# Test extract_text_response
good_response = {"content": [{"type": "text", "text": "Hello there"}], "stop_reason": "end_turn"}
assert extract_text_response(good_response) == "Hello there"
print("extract_text_response valid: PASS")
malformed_response = {"stop_reason": "end_turn"} # missing "content" key entirely
assert extract_text_response(malformed_response) == ""
print("extract_text_response malformed: PASS")Run test_prompt_utils.py — all tests, from Week P1 onward, should still pass. Commit this checkpoint before moving on:
git add prompt_utils.py test_prompt_utils.py && git commit -m "extend prompt_utils with request/response and logging helpers"
claude_client.py imports build_messages_array, extract_text_response, and append_jsonl directly (save_interaction_log is a standalone convenience function for scripts that don't need the full client — it isn't called from claude_client.py).
Create claude_client.py:
import os
import json
import logging
import requests
from datetime import datetime, timezone
from pathlib import Path
from dotenv import load_dotenv
from prompt_utils import build_messages_array, extract_text_response, append_jsonl
load_dotenv()
logger = logging.getLogger(__name__)
class ClaudeClient:
"""
A minimal Claude API client with credential management and interaction logging.
"""
def __init__(self, log_dir: str = "logs"):
self.api_key = os.environ.get("ANTHROPIC_API_KEY")
self.model = os.environ.get("ANTHROPIC_MODEL", "claude-haiku-4-5-20251001")
self.api_url = os.environ.get("ANTHROPIC_API_URL",
"https://api.anthropic.com/v1/messages")
if not self.api_key:
raise EnvironmentError("ANTHROPIC_API_KEY not set")
self.log_dir = Path(log_dir)
self.log_dir.mkdir(exist_ok=True)
self.log_file = self.log_dir / "interactions.jsonl"
def complete(self, system_prompt: str, user_message: str,
max_tokens: int = 1024, temperature: float = 1.0) -> dict:
"""
Send a single-turn completion request. Returns a result dict with
keys: success, response_text, input_tokens, output_tokens, error.
"""
headers = {
"x-api-key": self.api_key,
"anthropic-version": "2023-06-01",
"content-type": "application/json"
}
body = build_messages_array(system_prompt, user_message)
body["model"] = self.model
body["max_tokens"] = max_tokens
body["temperature"] = temperature
result = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"model": self.model,
"system_prompt_chars": len(system_prompt),
"user_message_chars": len(user_message)
}
try:
response = requests.post(self.api_url, headers=headers,
json=body, timeout=30)
response.raise_for_status()
data = response.json()
result["success"] = True
result["response_text"] = extract_text_response(data)
result["input_tokens"] = data["usage"]["input_tokens"]
result["output_tokens"] = data["usage"]["output_tokens"]
result["stop_reason"] = data["stop_reason"]
except requests.exceptions.HTTPError as e:
result["success"] = False
result["error"] = f"HTTP {e.response.status_code}: {e.response.text[:200]}"
logger.error("API HTTP error: %s", result["error"])
except requests.exceptions.Timeout:
result["success"] = False
result["error"] = "Request timed out after 30 seconds"
logger.error("API timeout")
except Exception as e:
result["success"] = False
result["error"] = str(e)
logger.error("Unexpected error: %s", str(e))
# Always log, success or failure
self._log_interaction(result)
return result
def _log_interaction(self, record: dict) -> None:
append_jsonl(record, str(self.log_file))Step 3 — Write a smoke test
Create test_claude_client.py. Send three requests: one that should succeed, one with an intentionally bad system prompt (very long, >200,000 chars), and one where you temporarily corrupt the API key to test error handling. Verify the log file is created and contains records for all three attempts, including the failures.
After the three requests, call save_interaction_log("smoke test", "3 cases: success/oversized-prompt/bad-key", "<summary of which passed>", "smoke_test_log.jsonl"). This is save_interaction_log's actual use case: a one-line log call from a standalone script that has no ClaudeClient instance to hang logging off of — ClaudeClient logs its own calls automatically via _log_interaction/append_jsonl, but a bare script summarising a test run has nothing else to do that job. Verify smoke_test_log.jsonl was created with your one record in it.
Step 4 — Commit
git add claude_client.py test_claude_client.py && git commit -m "add Claude API client with env credential loading and JSONL logging"
Reflection
Under a new ## Week P3 heading in journal.md, write 4–6 sentences answering the following. Be specific — reference the actual code you wrote, not the general concept.
The ClaudeClient you built this week stores the API key in self.api_key and the log file path in self.log_file. In a team of three engineers all using this client, what problem would arise if each engineer's .env file pointed to a different ANTHROPIC_MODEL? How does the environment variable pattern make that problem visible rather than silent? And: the _log_interaction method writes to a JSONL file — what would happen if two instances of ClaudeClient wrote to the same file concurrently in a multi-threaded service, and what is the architectural decision you would make to prevent it?
Commit the journal: git add journal.md && git commit -m "week P3 reflection"
Self-check questions
Q1. A Claude API response has the structure {"content": [{"type": "text", "text": "Hello"}], "stop_reason": "end_turn"}. Which Python expression correctly extracts the text string?
Q2. Your script loads ANTHROPIC_API_KEY from .env using python-dotenv. When this script is deployed to AWS Lambda, where should the API key value come from?
Q3. response.raise_for_status() will raise an exception for which HTTP status codes?
Q4. You need to store 10,000 interaction records in a file so that each record can be streamed and processed independently without loading the entire file into memory. Which format is correct?
Progression gate
No standalone progression-gate section in the source for this week — self-attestation still available below.