Hands-on 08-notifications-for-downloads-failures-and-request: Lab

scope

All created or modified files remain beneath /opt/lab-classroom/class63/. The lab performs no external network requests and sends no real notifications.

artifacts

/opt/lab-classroom/class63/notifier.py
/opt/lab-classroom/class63/events.jsonl
/opt/lab-classroom/class63/outbox.jsonl
/opt/lab-classroom/class63/deadletter.jsonl
/opt/lab-classroom/class63/state.json

steps

step

1

instruction

Create the dedicated lab directory.

command

mkdir -p /opt/lab-classroom/class63/

step

2

instruction

Create the local notification processor. It validates events, recursively redacts sensitive keys, routes recognized types, suppresses duplicates, and records malformed input by digest rather than raw content.

command

cat > /opt/lab-classroom/class63/notifier.py <<'PY'
import hashlib
import json
from pathlib import Path

BASE = Path('/opt/lab-classroom/class63')
EVENTS = BASE / 'events.jsonl'
OUTBOX = BASE / 'outbox.jsonl'
DEAD = BASE / 'deadletter.jsonl'
STATE = BASE / 'state.json'

POLICY = {
'download.completed': ('info', 'activity'),
'download.failed': ('critical', 'alerts'),
'request.created': ('info', 'activity'),
'request.approved': ('notice', 'requests'),
'request.denied': ('notice', 'requests'),
}
SENSITIVE_KEYS = {'token', 'password', 'api_key', 'authorization', 'secret'}

def redact(value):
if isinstance(value, dict):
return {
key: ('[REDACTED]' if key.lower() in SENSITIVE_KEYS else redact(item))
for key, item in value.items()
}
if isinstance(value, list):
return [redact(item) for item in value]
return value

def load_seen():
if not STATE.exists():
return set()
data = json.loads(STATE.read_text(encoding='utf-8'))
return set(data.get('seen', []))

def save_seen(seen):
temporary = BASE / 'state.tmp'
temporary.write_text(
json.dumps({'seen': sorted(seen)}, indent=2) + '\n',
encoding='utf-8'
)
temporary.replace(STATE)

def append_json(path, record):
with path.open('a', encoding='utf-8') as handle:
handle.write(json.dumps(record, sort_keys=True) + '\n')

def validate(event):
if not isinstance(event, dict):
raise ValueError('event must be a JSON object')
for field in ('event_id', 'type', 'timestamp', 'payload'):
if field not in event:
raise ValueError('missing required field: ' + field)
if not isinstance(event['event_id'], str) or not event['event_id'].strip():
raise ValueError('event_id must be a non-empty string')
if event['type'] not in POLICY:
raise ValueError('unsupported event type')
if not isinstance(event['timestamp'], str) or not event['timestamp'].strip():
raise ValueError('timestamp must be a non-empty string')
if not isinstance(event['payload'], dict):
raise ValueError('payload must be a JSON object')

def main():
BASE.mkdir(parents=True, exist_ok=True)
seen = load_seen()
queued = 0
duplicates = 0
rejected = 0

with EVENTS.open('r', encoding='utf-8') as source:
    for line_number, raw in enumerate(source, start=1):
        if not raw.strip():
            continue
        digest = hashlib.sha256(raw.encode('utf-8')).hexdigest()
        dead_key = 'dead:' + digest
        try:
            event = json.loads(raw)
            validate(event)
            event_key = 'event:' + event['event_id']
            if event_key in seen:
                duplicates += 1
                continue
            clean_payload = redact(event['payload'])
            severity, destination = POLICY[event['type']]
            notification_id = hashlib.sha256(
                (event['event_id'] + ':' + event['type']).encode('utf-8')
            ).hexdigest()
            notification = {
                'notification_id': notification_id,
                'event_id': event['event_id'],
                'event_type': event['type'],
                'event_timestamp': event['timestamp'],
                'severity': severity,
                'destination': destination,
                'subject': event['type'].replace('.', ' ').title(),
                'payload': clean_payload,
                'delivery_status': 'queued'
            }
            append_json(OUTBOX, notification)
            seen.add(event_key)
            save_seen(seen)
            queued += 1
        except Exception as exc:
            if dead_key not in seen:
                append_json(DEAD, {
                    'line_number': line_number,
                    'raw_sha256': digest,
                    'error': str(exc)
                })
                seen.add(dead_key)
                save_seen(seen)
                rejected += 1
            else:
                duplicates += 1

print(json.dumps({
    'queued': queued,
    'duplicates_or_known_rejections': duplicates,
    'newly_rejected': rejected
}, sort_keys=True))

if name == 'main':
main()
PY

step

3

instruction

Create four valid sample events and one malformed event. The failed-download event deliberately contains a sample secret value so redaction can be verified.

command

cat > /opt/lab-classroom/class63/events.jsonl <<'JSONL'
{"event_id":"evt-001","type":"download.completed","timestamp":"2025-03-08T10:00:00Z","payload":{"item":"Example Documentary","quality":"1080p"}}
{"event_id":"evt-002","type":"download.failed","timestamp":"2025-03-08T10:05:00Z","payload":{"item":"Example Series S01E01","reason":"provider timeout","api_key":"LAB-SECRET-VALUE"}}
{"event_id":"evt-003","type":"request.created","timestamp":"2025-03-08T10:10:00Z","payload":{"item":"Example Film","request_id":42}}
{"event_id":"evt-004","type":"request.approved","timestamp":"2025-03-08T10:15:00Z","payload":{"item":"Example Film","request_id":42}}
{"type":"download.failed","timestamp":"2025-03-08T10:20:00Z","payload":{"reason":"missing event identifier"}}
JSONL

step

4

instruction

Run the processor for the first time. Four records should be queued and one malformed record should be rejected.

command

python3 /opt/lab-classroom/class63/notifier.py

step

5

instruction

Run the processor again to test idempotency. No additional outbox or dead-letter records should be created.

command

python3 /opt/lab-classroom/class63/notifier.py

step

6

instruction

Inspect the generated artifacts without contacting an external service.

command

python3 - <<'PY'
import json
from pathlib import Path
base = Path('/opt/lab-classroom/class63')
for name in ('outbox.jsonl', 'deadletter.jsonl', 'state.json'):
path = base / name
print('\n==', name, '==')
print(path.read_text(encoding='utf-8'))
PY

step

7

instruction

Perform deterministic assertions for record counts, routing, status, deduplication, and redaction.

command

python3 - <<'PY'
import json
from pathlib import Path
base = Path('/opt/lab-classroom/class63')
outbox = [json.loads(line) for line in (base / 'outbox.jsonl').read_text(encoding='utf-8').splitlines() if line]
dead = [json.loads(line) for line in (base / 'deadletter.jsonl').read_text(encoding='utf-8').splitlines() if line]
state = json.loads((base / 'state.json').read_text(encoding='utf-8'))
assert len(outbox) == 4, outbox
assert len(dead) == 1, dead
assert len(state['seen']) == 5, state
assert all(item['delivery_status'] == 'queued' for item in outbox)
failed = next(item for item in outbox if item['event_type'] == 'download.failed')
assert failed['severity'] == 'critical'
assert failed['destination'] == 'alerts'
assert failed['payload']['api_key'] == '[REDACTED]'
serialized = json.dumps(outbox)
assert 'LAB-SECRET-VALUE' not in serialized
assert 'LAB-SECRET-VALUE' not in json.dumps(dead)
assert set(item['event_id'] for item in outbox) == {'evt-001', 'evt-002', 'evt-003', 'evt-004'}
print('PASS: routing, deduplication, dead-letter handling, and redaction verified')
PY

production_extension

Replace the demonstration file producer with authenticated events from approved applications.
Keep normalization and policy separate from provider-specific delivery code.
Add an outbox delivery worker with explicit timeouts, bounded retries, increasing delays, and random jitter.
Record delivery attempts and terminal failures without writing provider credentials into logs.
Use provider-side identifiers or idempotency keys when supported.
Test provider failure behavior in a non-production destination before enabling household or operations channels.