Architecture starting code
These files are small, runnable mechanism demonstrations, not complete services. Each corresponding project page explains the application, the local assignment and the code/adapters needed to reach its AWS diagram. Running one of these programs creates no cloud resources.
For an actual HTTP application to extend, start with the reading-list API and request walkthrough. For the durable link-monitor CLI, use the link-watcher reference. The AI workflow code on GitHub contains four local workflows with fixture providers.
Download the repository once:
git clone https://github.com/Soulful-Iris/junior-to-staff.git
cd junior-to-staff
Then run the command in the project's page. Python 3.12+ is sufficient for these standard-library demonstrations. The printed output is a first observation. Completing the assignment requires the additional implementation and failure walkthroughs named on that page.
Read the supplied code · bookmark_service.py
"""Local mechanism demonstration for bookmark-service. No AWS resources are created."""
import sqlite3
c=sqlite3.connect(':memory:')
c.execute('CREATE TABLE bookmarks(owner TEXT,id INTEGER,title TEXT,version INTEGER,PRIMARY KEY(owner,id))')
c.execute("INSERT INTO bookmarks VALUES ('ana',7,'Original',3)")
for owner,title,version in [('ben','Unauthorized',3),('ana','Phone edit',3),('ana','Laptop edit',3)]:
changed=c.execute('UPDATE bookmarks SET title=?,version=version+1 WHERE owner=? AND id=7 AND version=?',(title,owner,version)).rowcount
print(owner,title,'saved' if changed else 'not found or version conflict')
print('Final:',c.execute('SELECT * FROM bookmarks').fetchall())
Read the supplied code · url_shortener.py
"""Local mechanism demonstration for url-shortener. No AWS resources are created."""
import sqlite3
c=sqlite3.connect(':memory:')
c.execute('CREATE TABLE links(code TEXT PRIMARY KEY,target TEXT,expires INTEGER)')
for target in ('https://example.com/one','https://example.com/two'):
try:
c.execute('INSERT INTO links VALUES (?,?,?)',('launch',target,100))
print('201 reserved launch for',target)
except sqlite3.IntegrityError:
print('409 alias already owned')
cached=c.execute('SELECT target,expires FROM links WHERE code=?',('launch',)).fetchone()
for now in (99,100,101):
print('now',now,('302 '+cached[0]) if now<cached[1] else '410 expired')
Read the supplied code · api_quota.py
"""Local mechanism demonstration for api-quota. No AWS resources are created."""
import sqlite3
c=sqlite3.connect(':memory:')
c.execute('CREATE TABLE quota(org TEXT PRIMARY KEY,remaining INTEGER CHECK(remaining>=0))')
c.execute("INSERT INTO quota VALUES ('acme',1)")
for gateway in ('gateway-a','gateway-b'):
allowed=c.execute('UPDATE quota SET remaining=remaining-1 WHERE org=? AND remaining>0',('acme',)).rowcount
print(gateway,'allowed' if allowed else '429 quota exhausted')
print('remaining:',c.execute('SELECT remaining FROM quota').fetchone()[0])
Read the supplied code · checkout_payment.py
"""Local mechanism demonstration for checkout-payment. No AWS resources are created."""
# A tiny provider fixture models a charge whose response was lost.
provider={}; order={'id':'order-7','state':'pending'}
key='payment:order-7:v1'
def charge(k):
if k not in provider: provider[k]={'charge_id':'charge-91','amount_minor':1999}
return provider[k]
charge(key) # provider committed; simulate losing the response
order['state']='payment_unknown'
print('After timeout:',order)
evidence=charge(key) # the SAME provider key resolves the same charge
order.update(state='paid',charge_id=evidence['charge_id'])
print('After reconciliation:',order,'provider charge count:',len(provider))
Read the supplied code · audit_trail.py
"""Local mechanism demonstration for audit-trail. No AWS resources are created."""
import sqlite3,json
c=sqlite3.connect(':memory:')
c.executescript('CREATE TABLE permissions(id TEXT PRIMARY KEY,version INT,access TEXT); CREATE TABLE outbox(id TEXT PRIMARY KEY,payload TEXT);')
c.execute("INSERT INTO permissions VALUES ('report-7',1,'private')"); c.commit()
with c:
c.execute("UPDATE permissions SET version=2,access='team' WHERE id='report-7' AND version=1")
c.execute('INSERT INTO outbox VALUES (?,?)',('report-7:2',json.dumps({'actor':'ana','before':'private','after':'team'})))
print('After simulated crash before export:',c.execute('SELECT * FROM permissions').fetchall())
print('Still available to relay:',c.execute('SELECT * FROM outbox').fetchall())
Read the supplied code · tenant_isolation.py
"""Local mechanism demonstration for tenant-isolation. No AWS resources are created."""
reports={('acme','r1'):'Acme revenue',('birch','r2'):'Birch payroll'}
members={('ana','acme')}; cache={('acme','r1'):'Acme revenue'}
def read(user,tenant,rid):
if (user,tenant) not in members: return '403 membership revoked'
return cache.get((tenant,rid),reports.get((tenant,rid),'404 scoped resource'))
print(read('ana','acme','r2'))
print(read('ana','acme','r1'))
members.remove(('ana','acme'))
print(read('ana','acme','r1'))
Read the supplied code · durable_jobs.py
"""Local mechanism demonstration for durable-jobs. No AWS resources are created."""
job={'epoch':4,'state':'running','result':None}; objects={}
def publish(epoch,key):
if job['epoch']!=epoch or job['state']!='running': return 'stale publication rejected'
job.update(state='succeeded',result=key); return 'published'
old_epoch=4
job['epoch']=5
objects['job7/5/result']='new worker bytes'
print(publish(5,'job7/5/result'))
objects['job7/4/result']='late old worker bytes'
print(publish(old_epoch,'job7/4/result'))
print('Visible result:',job['result'],objects[job['result']])
Read the supplied code · job_scheduler.py
"""Local mechanism demonstration for job-scheduler. No AWS resources are created."""
import sqlite3
c=sqlite3.connect(':memory:')
c.execute('CREATE TABLE runs(schedule TEXT,version INT,due TEXT,state TEXT,PRIMARY KEY(schedule,version,due))')
for dispatcher in ('old-dispatcher','new-dispatcher'):
changed=c.execute('INSERT OR IGNORE INTO runs VALUES (?,?,?,?)',('report-7',2,'2026-01-10T09:00:00Z','accepted')).rowcount
print(dispatcher,'created run' if changed else 'existing run')
print(c.execute('SELECT * FROM runs').fetchall())
Read the supplied code · erasure_workflow.py
"""Local mechanism demonstration for erasure-workflow. No AWS resources are created."""
ledger={'ana':6}; records={'ana':{'version':5,'email':'ana@example.invalid'}}
def import_record(subject,version,payload):
if version<=ledger.get(subject,-1): return 'rejected by deletion generation'
records[subject]=payload; return 'imported'
records.pop('ana',None)
print(import_record('ana',4,{'email':'old backup copy'}))
print('Readable account:',records.get('ana','absent'))
print('Deletion evidence:',ledger)
Read the supplied code · multi_tenant_migration.py
"""Local mechanism demonstration for multi-tenant-migration. No AWS resources are created."""
target={}
def apply(key,version,value):
old=target.get(key)
if old and old['version']>=version: return 'ignored stale/duplicate'
target[key]={'version':version,'value':value}; return 'applied'
for v,value in [(5,'live edit'),(6,None),(4,'old backfill')]:
print(v,apply('acme:7',v,value))
print('Target:',target,'visible:',target['acme:7']['value'])
Read the supplied code · regional_failover.py
"""Local mechanism demonstration for regional-failover. No AWS resources are created."""
authority={'epoch':7,'region':'east'}; orders=[]
def commit(region,epoch,order):
if (region,epoch)!=(authority['region'],authority['epoch']): return 'fenced'
orders.append(order); return 'committed'
print(commit('east',7,'order-1'))
authority.update(epoch=8,region='west')
print(commit('east',7,'late-order'))
print(commit('west',8,'order-2'))
print(orders)
Read the supplied code · calendar_availability.py
"""Local mechanism demonstration for calendar-availability. No AWS resources are created."""
bookings=[]
def reserve(start,end):
if end<=start: return '400 invalid interval'
if any(start<b and a<end for a,b in bookings): return '409 room occupied'
bookings.append((start,end)); return '201 reserved'
for interval in [(600,660),(630,690),(660,720)]:
print(interval,reserve(*interval))
print('Minutes from midnight:',bookings)
Read the supplied code · ticket_inventory.py
"""Local mechanism demonstration for ticket-inventory. No AWS resources are created."""
seat={'owner':None,'expires':0,'sold':False}
def hold(owner,now):
if seat['sold'] or (seat['owner'] and now<seat['expires']): return '409 unavailable'
seat.update(owner=owner,expires=now+300); return 'held'
def sell(owner,now):
if seat['owner']!=owner or now>=seat['expires'] or seat['sold']: return '409 hold invalid'
seat['sold']=True; return 'sold'
print('Ana',hold('ana',0)); print('Ben',hold('ben',1))
print('Ben after expiry',hold('ben',301)); print('Late Ana',sell('ana',302))
print('Ben',sell('ben',302))
Read the supplied code · rideshare_dispatch.py
"""Local mechanism demonstration for rideshare-dispatch. No AWS resources are created."""
rides={'r1':None,'r2':None}; drivers={'d7':None}
def accept(ride,driver):
if rides[ride] is not None or drivers[driver] is not None: return '409 already assigned'
rides[ride]=driver; drivers[driver]=ride; return 'assignment committed'
print(accept('r1','d7')); print(accept('r2','d7')); print(rides,drivers)
latest={'seq':9,'cell':'north'}
for seq,cell in [(8,'south'),(10,'east')]:
if seq>latest['seq']: latest.update(seq=seq,cell=cell)
print('Latest location:',latest)
Read the supplied code · realtime_chat.py
"""Local mechanism demonstration for realtime-chat. No AWS resources are created."""
messages=[]; operations={}
def send(sender,key,text):
identity=(sender,key)
if identity in operations: return operations[identity]
record={'sequence':len(messages)+1,'sender':sender,'text':text}
messages.append(record); operations[identity]=record; return record
print(send('ana','phone-7','On my way'))
print(send('ana','phone-7','On my way'))
print('Reconnect after sequence 0:',[m for m in messages if m['sequence']>0])
Read the supplied code · collaborative_editor.py
"""Local mechanism demonstration for collaborative-editor. No AWS resources are created."""
doc={'revision':12,'text':'Incident draft'}; accepted={}
def edit(op,base,text):
if op in accepted: return accepted[op]
if base!=doc['revision']: return {'status':409,'server_revision':doc['revision'],'keep_draft':text}
doc.update(revision=base+1,text=text); accepted[op]=dict(doc); return accepted[op]
print(edit('ana-1',12,'Ana update'))
print(edit('ben-1',12,'Ben offline draft'))
print(edit('ana-1',12,'Ana update'))
Read the supplied code · notification_platform.py
"""Local mechanism demonstration for notification-platform. No AWS resources are created."""
preferences={'ana':True}; intents={}; attempts=[]
def queue(event,user): intents.setdefault((event,user),{'state':'queued'})
def deliver(event,user):
intent=intents[(event,user)]
if not preferences.get(user,False): intent['state']='suppressed'; return
if intent['state']=='provider_accepted': return
attempts.append((event,user)); intent['state']='provider_accepted'
queue('campaign-7','ana'); queue('campaign-7','ana')
preferences['ana']=False; deliver('campaign-7','ana')
print(intents,'provider attempts:',attempts)
Read the supplied code · webhook_delivery.py
"""Local mechanism demonstration for webhook-delivery. No AWS resources are created."""
import hmac,hashlib,json
body=json.dumps({'id':'evt-7','type':'order.paid'},separators=(',',':')).encode()
secret=b'local-fixture-secret'; timestamp='1700000000'
signature=hmac.new(secret,timestamp.encode()+b'.'+body,hashlib.sha256).hexdigest()
print('Signed bytes:',body.decode()); print('Signature:',signature)
seen=set()
for attempt in (1,2):
event=json.loads(body)['id']; duplicate=event in seen; seen.add(event)
print('attempt',attempt,'already handled' if duplicate else 'apply event')
Read the supplied code · food_delivery_marketplace.py
"""Local mechanism demonstration for food-delivery-marketplace. No AWS resources are created."""
menu={'meal7':{'price':1250,'version':3,'available':True}}
cart={'meal7':{'price':1250,'version':3}}
menu['meal7'].update(price=1400,version=4)
def checkout(cart):
for item,seen in cart.items():
current=menu[item]
if not current['available']: return '409 item unavailable'
if seen['version']!=current['version']: return {'status':409,'new_price':current['price'],'needs_acceptance':True}
return 'order accepted'
print(checkout(cart))
Read the supplied code · social_feed.py
"""Local mechanism demonstration for social-feed. No AWS resources are created."""
posts={'p1':{'text':'Team update','public':True},'p2':{'text':'Old private draft','public':False}}
feed=['p2','p1']; deleted={'p3'}
def visible(ids):
return [posts[i]['text'] for i in ids if i not in deleted and i in posts and posts[i]['public']]
print('Warm feed candidates:',feed); print('Returned content:',visible(feed))
posts['p1']['public']=False
print('After revocation:',visible(feed))
Read the supplied code · news_aggregator.py
"""Local mechanism demonstration for news-aggregator. No AWS resources are created."""
articles={}; etag='v1'
def ingest(source,item,url,title):
key=(source,item); articles[key]={'url':url,'title':title}
return 'upserted '+str(key)
print(ingest('publisher-a','42','https://example.invalid/story','First title'))
print(ingest('publisher-a','42','https://example.invalid/story','Corrected title'))
print('304 response: retain existing articles',articles)
Read the supplied code · video_processing.py
"""Local mechanism demonstration for video-processing. No AWS resources are created."""
required={'360p','720p','1080p'}; outputs={}; published=None
def publish():
global published
if not required.issubset(outputs): return 'still processing: '+str(sorted(required-outputs.keys()))
published=dict(outputs); return 'ready'
outputs.update({'360p':'g2/360.mp4','720p':'g2/720.mp4'})
print(publish()); outputs['1080p']='g2/1080.mp4'; print(publish()); print(published)
Read the supplied code · video_streaming_platform.py
"""Local mechanism demonstration for video-streaming-platform. No AWS resources are created."""
session={'expected_parts':{1,2,3},'received':{1:'hash-a',3:'hash-c'}}
print('Resume missing parts:',sorted(session['expected_parts']-session['received'].keys()))
session['received'][2]='hash-b'
print('Upload complete:',session['expected_parts']==set(session['received']))
video={'ready':True,'private':True}; members={'ana'}
for viewer in ('ana','ben'):
print(viewer,'manifest allowed' if video['ready'] and (not video['private'] or viewer in members) else '403')
Read the supplied code · online_judge.py
"""Local mechanism demonstration for online-judge. No AWS resources are created."""
import subprocess,sys
program="while True: pass"
try:
subprocess.run([sys.executable,'-c',program],timeout=0.2,capture_output=True)
except subprocess.TimeoutExpired:
print('wall-clock limit exceeded; child terminated')
print('This subprocess example is NOT a security sandbox.')
Read the supplied code · api_gateway_platform.py
"""Local mechanism demonstration for api-gateway-platform. No AWS resources are created."""
active={'version':1,'routes':{'/orders':'orders-v1'}}
def activate(candidate):
if not candidate.get('routes') or any(not p.startswith('/') for p in candidate['routes']): return 'rejected; keep version '+str(active['version'])
active.clear(); active.update(candidate); return 'active version '+str(active['version'])
print(activate({'version':2,'routes':{'orders':'broken'}}))
print(activate({'version':3,'routes':{'/orders':'orders-v2'}}))
print(active)
Read the supplied code · event_ingestion.py
"""Local mechanism demonstration for event-ingestion. No AWS resources are created."""
events=[{'id':'e1','version':1,'value':7},{'id':'e1','version':1,'value':7},{'id':'e2','version':99,'value':8}]
seen=set(); output=[]; quarantine=[]
for event in events:
if event['version']!=1: quarantine.append((event,'unsupported schema')); continue
if event['id'] in seen: continue
output.append(event['value']); seen.add(event['id'])
print('Applied:',output,'quarantine:',quarantine)
Read the supplied code · ad_click_aggregator.py
"""Local mechanism demonstration for ad-click-aggregator. No AWS resources are created."""
seen=set(); windows={}; watermark=120; lateness=120
for eid,campaign,at in [('e1','c7',61),('e1','c7',61),('e2','c7',10),('e3','c7',-180)]:
if eid in seen: print(eid,'duplicate'); continue
if at<watermark-lateness: print(eid,'too late: reconciliation path'); continue
seen.add(eid); key=(campaign,(at//60)*60); windows[key]=windows.get(key,0)+1
print('Provisional counts:',windows)
Read the supplied code · trending_counts.py
"""Local mechanism demonstration for trending-counts. No AWS resources are created."""
import hashlib
seen=set(); partials={}
for eid,topic in [('e1','goal'),('e2','goal'),('e1','goal'),('e3','save')]:
if eid in seen: continue
seen.add(eid); shard=int(hashlib.sha256(eid.encode()).hexdigest(),16)%4
key=(topic,shard); partials[key]=partials.get(key,0)+1
totals={}
for (topic,shard),count in partials.items(): totals[topic]=totals.get(topic,0)+count
print('Partials:',partials); print('Ranking:',sorted(totals.items(),key=lambda x:(-x[1],x[0])))
Read the supplied code · distributed_cache.py
"""Local mechanism demonstration for distributed-cache. No AWS resources are created."""
cache={'item7':{'version':5,'value':'new price'}}
def fill(key,version,value):
current=cache.get(key)
if current and current['version']>version: return 'stale fill rejected'
cache[key]={'version':version,'value':value}; return 'filled'
print(fill('item7',4,'old price')); print(cache)
requests=1000; per_process_coalesced=1; processes=10
print('One hot miss:',requests,'requests can still cause',processes*per_process_coalesced,'origin reads across processes')
Read the supplied code · distributed_key_value_store.py
"""Local mechanism demonstration for distributed-key-value-store. No AWS resources are created."""
import json,os,tempfile
with tempfile.TemporaryDirectory() as directory:
path=directory+'/wal.jsonl'
with open(path,'w') as log:
log.write(json.dumps({'index':1,'term':3,'key':'a','value':'saved'})+'\n')
log.flush(); os.fsync(log.fileno())
recovered={}
with open(path) as log:
for line in log:
entry=json.loads(line); recovered[entry['key']]=entry['value']
print('Recovered after closing the writer:',recovered)
print('Local WAL only: this is not a replicated consensus implementation.')
Read the supplied code · document_search.py
"""Local mechanism demonstration for document-search. No AWS resources are created."""
index=[{'id':'public','score':9},{'id':'secret','score':10}]
content={'public':'Engineering handbook','secret':'Private acquisition plan'}
allowed={'public'}
results=[]
for hit in sorted(index,key=lambda h:-h['score']):
if hit['id'] not in allowed: continue
results.append({'id':hit['id'],'snippet':content[hit['id']]})
print(results)
Read the supplied code · typeahead_search.py
"""Local mechanism demonstration for typeahead-search. No AWS resources are created."""
latest={'generation':2,'query':'iph'}
responses=[{'generation':2,'query':'iph','items':['iphone']},{'generation':1,'query':'ipa','items':['ipad']}]
for response in responses:
print('display' if response['generation']==latest['generation'] else 'ignore stale',response['items'])
index={'iph':['iphone','iphone case','iphone charger']}
print('Suggestions:',index.get('iph',[])[:5])
Read the supplied code · file_synchronization.py
"""Local mechanism demonstration for file-synchronization. No AWS resources are created."""
file={'version':3,'manifest':'hash-original','deleted':False}
def publish(base,manifest):
if base!=file['version']: return {'status':409,'keep_local':manifest,'server':dict(file)}
file.update(version=base+1,manifest=manifest); return dict(file)
print(publish(3,'hash-ana'))
print(publish(3,'hash-ben'))
file.update(version=5,deleted=True,manifest=None)
print(publish(3,'hash-old-laptop'))
Read the supplied code · web_crawler.py
"""Local mechanism demonstration for web-crawler. No AWS resources are created."""
from urllib.parse import urlsplit,urlunsplit
seen=set()
def normalize(url):
p=urlsplit(url)
if p.scheme not in ('http','https') or p.username or p.password: raise ValueError('unsupported URL')
return urlunsplit((p.scheme.lower(),p.netloc.lower(),p.path or '/',p.query,''))
for url in ['https://EXAMPLE.invalid/a#one','https://example.invalid/a#two','https://example.invalid/a?sort=1']:
key=normalize(url); print(key,'duplicate' if key in seen else 'new'); seen.add(key)
print('Normalization only; network destination validation is a separate required boundary.')
Read the supplied code · feature_rollout.py
"""Local mechanism demonstration for feature-rollout. No AWS resources are created."""
import hashlib
def exposed(subject,percent):
bucket=int(hashlib.sha256(('invoice-v1:'+subject).encode()).hexdigest(),16)%10000
return bucket<percent*100
users=[f'user-{i}' for i in range(200)]
users=[u for u in users if exposed(u,5)][:2]+[u for u in users if exposed(u,25) and not exposed(u,5)][:2]
for user in users:
print(user,'5%',exposed(user,5),'25%',exposed(user,25),'repeat',exposed(user,5))
print('Emergency off:',[exposed(u,0) for u in ['ana','ben','chen']])
Read the supplied code · slow_request.py
"""Local mechanism demonstration for slow-request. No AWS resources are created."""
import math
samples=[80]*96+[1900]*4
samples.sort()
for percentile in (50,95,99):
value=samples[math.ceil(percentile/100*len(samples))-1]
print('p'+str(percentile),value,'ms')
trace={'pool_wait_ms':1700,'query_ms':40,'application_ms':160}
print('Slow request phases:',trace,'total',sum(trace.values()),'ms')
Read the supplied code · metrics_platform.py
"""Local mechanism demonstration for metrics-platform. No AWS resources are created."""
buckets=[{'count':1,'sum':100},{'count':9,'sum':90}]
wrong=sum(b['sum']/b['count'] for b in buckets)/len(buckets)
correct=sum(b['sum'] for b in buckets)/sum(b['count'] for b in buckets)
print('Unweighted average of averages:',wrong)
print('Correct weighted average:',correct)
labels={'service':'checkout','request_id':'unique-123'}
print('Rejected labels:',sorted(set(labels)-{'service','region','status'}))
Read the supplied code · overload_shedding.py
"""Local mechanism demonstration for overload-shedding. No AWS resources are created."""
arrival=40000; service=20000; limit=100000; queued=0
for second in range(1,8):
offered=queued+arrival; completed=min(service,offered)
waiting=offered-completed; rejected=max(0,waiting-limit); queued=min(waiting,limit)
print(second,'completed',completed,'queued',queued,'rejected',rejected)
Read the supplied code · knowledge_assistant.py
"""Local mechanism demonstration for knowledge-assistant. No AWS resources are created."""
docs={'d1':{'text':'Expenses require a receipt.','allowed':True},'d2':{'text':'Secret acquisition plan','allowed':False}}
candidates=['d2','d1']; evidence=[(i,docs[i]['text']) for i in candidates if docs[i]['allowed']]
print('Authorized evidence:',evidence)
docs['d1']['allowed']=False
print('Final response:', 'withheld: access changed' if any(not docs[i]['allowed'] for i,_ in evidence) else evidence)
Read the supplied code · support_assistant.py
"""Local mechanism demonstration for support-assistant. No AWS resources are created."""
import hashlib,json
proposal={'type':'credit','ticket':'t7','amount_minor':500,'currency':'USD'}
def digest(p): return hashlib.sha256(json.dumps(p,sort_keys=True).encode()).hexdigest()
approved=digest(proposal)
proposal['amount_minor']=5000
print('Execution allowed:',digest(proposal)==approved)
print('Changed parameters require a new approval.')
Read the supplied code · personalized_ranking.py
"""Local mechanism demonstration for personalized-ranking. No AWS resources are created."""
candidates=[{'id':'a','score':9,'eligible':False},{'id':'b','score':5,'eligible':True},{'id':'c','score':3,'eligible':True}]
feature_elapsed_ms=170; remaining_ms=250-feature_elapsed_ms
mode='fallback' if remaining_ms<80+20+30 else 'personalized'
ranked=sorted(candidates,key=lambda x:-x['score'])
print('Mode:',mode,'remaining budget:',remaining_ms)
print('Returned:',[x['id'] for x in ranked if x['eligible']])
Read the supplied code · 01_the_request_you_can_trace_end_to_end.py
"""Local mechanism demonstration for 01-the-request-you-can-trace-end-to-end. No AWS resources are created."""
import asyncio,contextvars,json
request_id=contextvars.ContextVar('request_id')
async def handle(name,slow):
token=request_id.set(name)
try:
for event in ('auth.ok','bookmark.saved','title.timeout' if slow else 'title.ok'):
await asyncio.sleep(0)
print(json.dumps({'request_id':request_id.get(),'event':event}))
finally: request_id.reset(token)
async def main(): await asyncio.gather(handle('A',True),handle('B',False))
asyncio.run(main())
Read the supplied code · 02_the_three_second_budget.py
"""Local mechanism demonstration for 02-the-three-second-budget. No AWS resources are created."""
budget=3000; spent=300; reserve=100
for attempt in (1,2):
remaining=budget-spent-reserve
timeout=min(1200,remaining)
if timeout<=0: print('no budget for attempt',attempt); break
print('attempt',attempt,'timeout',timeout,'ms')
spent+=timeout
if attempt==1: spent+=200
print('elapsed with response reserve:',spent+reserve,'ms')
Read the supplied code · 03_the_fetch_that_cannot_be_aimed_inward.py
"""Local mechanism demonstration for 03-the-fetch-that-cannot-be-aimed-inward. No AWS resources are created."""
import ipaddress
from urllib.parse import urlsplit
def permitted(url,resolved):
p=urlsplit(url)
if p.scheme not in ('http','https') or p.username or p.password or p.port not in (None,80,443): return False
return bool(resolved) and all(ipaddress.ip_address(a).is_global for a in resolved)
for url,addresses in [('https://example.com',['93.184.216.34']),('http://metadata.invalid',['169.254.169.254']),('https://mixed.invalid',['93.184.216.34','127.0.0.1'])]:
print(url,permitted(url,addresses))
print('Policy demonstration only; the network adapter must pin the validated IP.')
Read the supplied code · 04_the_api_that_does_not_break_its_callers.py
"""Local mechanism demonstration for 04-the-api-that-does-not-break-its-callers. No AWS resources are created."""
tags=[{'id':'t7','label':'databases','color':'blue'}]
def response():
return {'tags':[t['label'] for t in tags],'tagObjects':[dict(t) for t in tags]}
body=response(); print(body)
print('Old client:',', '.join(body['tags']))
print('New client:',body['tagObjects'][0]['id'])
Read the supplied code · 05_the_job_that_survives_a_restart.py
"""Local mechanism demonstration for 05-the-job-that-survives-a-restart. No AWS resources are created."""
job={'epoch':1,'state':'running','title':None}
def finish(epoch,title):
if job['epoch']!=epoch or job['state']!='running': return 'stale worker rejected'
job.update(state='succeeded',title=title); return 'saved'
old=job['epoch']; job['epoch']=2
print(finish(2,'Current title')); print(finish(old,'Late old title')); print(job)
Read the supplied code · a_public_form.py
"""Local mechanism demonstration for a-public-form. No AWS resources are created."""
import csv,io
rows=[['Ana','00123','=1+1\nPlease call after six']]
def spreadsheet_text(value):
return "'"+value if value.lstrip().startswith(('=','+','-','@')) else value
out=io.StringIO(newline=''); writer=csv.writer(out)
writer.writerow(['name','phone','notes'])
for row in rows: writer.writerow([row[0],"'"+row[1],spreadsheet_text(row[2])])
print(out.getvalue()); print('Stored originals:',rows)
Read the supplied code · a_receipt_tracker.py
"""Local mechanism demonstration for a-receipt-tracker. No AWS resources are created."""
from decimal import Decimal
receipt={'suggested_minor':19990,'confirmed_minor':None,'revision':1}
confirmed=int(Decimal('19.99')*100)
receipt.update(confirmed_minor=confirmed,revision=2)
receipt['suggested_minor']=19990
print('Machine suggestion:',receipt['suggested_minor'])
print('Report amount:',receipt['confirmed_minor'],'minor units; revision',receipt['revision'])
Read the supplied code · a_shared_reading_list.py
"""Local mechanism demonstration for a-shared-reading-list. No AWS resources are created."""
link={'version':3,'note':'Original'}; read_state={('ana','l7'):True,('ben','l7'):False}
def save(base,draft):
if base!=link['version']: return {'state':'conflict','draft':draft,'server':dict(link)}
link.update(version=base+1,note=draft); return {'state':'saved','server':dict(link)}
print(save(3,'Ana note')); print(save(3,'Ben draft')); print('Personal read state:',read_state)
Read the supplied code · a_shift_schedule.py
"""Local mechanism demonstration for a-shift-schedule. No AWS resources are created."""
shifts=[('ana',9,12)]; managers={'m1'}
def assign(actor,employee,start,end):
if actor not in managers: return '403 revoked'
if any(e==employee and start<b and a<end for e,a,b in shifts): return '409 overlapping shift'
shifts.append((employee,start,end)); return 'saved'
print(assign('m1','ana',11,14)); print(assign('m1','ana',12,15))
managers.remove('m1'); print(assign('m1','ben',9,12))
Read the supplied code · 01_the_suite_that_can_fail.py
"""Local mechanism demonstration for 01-the-suite-that-can-fail. No AWS resources are created."""
records=[('owner filter','semantic','detected'),('expiry boundary','semantic','survived'),('comment','equivalent','excluded'),('local rename','equivalent','excluded')]
relevant=[r for r in records if r[1]=='semantic']
print('Selected semantic defects:',len(relevant))
print('Detected:',sum(r[2]=='detected' for r in relevant))
print('Needs investigation:',[r[0] for r in relevant if r[2]=='survived'])
Read the supplied code · 02_the_contract_nobody_breaks_by_accident.py
"""Local mechanism demonstration for 02-the-contract-nobody-breaks-by-accident. No AWS resources are created."""
def decode(body):
duration=body.get('durationMs')
if not isinstance(body.get('title'),str): return 'invalid title'
if isinstance(duration,bool) or not isinstance(duration,(int,float)) or duration<0: return 'invalid durationMs'
return 'shape accepted; unit meaning still requires the contract'
for body in [{'title':'Guide','durationMs':12},{'title':'Guide','durationMs':'12'}]: print(body,decode(body))
Read the supplied code · 03_the_flake_hunter.py
"""Local mechanism demonstration for 03-the-flake-hunter. No AWS resources are created."""
def run(order,isolated):
shared=[]; outcomes=[]
for action in order:
store=[] if isolated else shared
if action=='create': store.append('item7'); outcomes.append('created')
else: outcomes.append('empty' if not store else 'unexpected leftover')
return outcomes
for isolated in (False,True):
print('isolated',isolated,run(['create','expect-empty'],isolated),run(['expect-empty','create'],isolated))
Read the supplied code · 04_the_load_test_that_finds_the_real_limit.py
"""Local mechanism demonstration for 04-the-load-test-that-finds-the-real-limit. No AWS resources are created."""
queue=0; arrivals=completed=0
for second in range(10):
arrivals+=120; queue+=120
done=min(100,queue); queue-=done; completed+=done
print({'arrivals':arrivals,'completed':completed,'waiting':queue})
print('A closed-loop generator changes offered rate when responses slow.')
Read the supplied code · 05_the_test_that_runs_in_production_forever.py
"""Local mechanism demonstration for 05-the-test-that-runs-in-production-forever. No AWS resources are created."""
steps={'create':'failed','read':'skipped','delete':'cleanup_attempted','verify_absent':'unknown'}
required=('create','read','delete','verify_absent')
print('Journey successful:',all(steps[s]=='passed' for s in required))
print('Step evidence:',steps)
last_completed=0; now=181; interval=60
print('Runner stale:',now-last_completed>3*interval)
Read the supplied code · twenty_deploys_a_day.py
"""Local mechanism demonstration for twenty-deploys-a-day. No AWS resources are created."""
releases={'v1':{'reads':{'title','url'}},'v2':{'reads':{'title','url','tagObjects'}}}
record={'title':'Guide','url':'https://example.invalid','tagObjects':[]}
active='v2'; flag=False
print('Deployed:',active,'feature enabled:',flag)
active='v1'
print('Rolled back:',active,'can read required fields:',releases[active]['reads'].issubset(record))
del record['title']
print('After destructive schema change:',releases[active]['reads'].issubset(record))
Read the supplied code · debuggable_at_three_in_the_morning.py
"""Local mechanism demonstration for debuggable-at-three-in-the-morning. No AWS resources are created."""
impact=0; metric=30; alert=70; mitigation=130
print({'collection_s':metric-impact,'alert_delivery_and_evaluation_s':alert-metric,'detection_s':alert-impact,'mitigation_after_alert_s':mitigation-alert})
events=[{'request':'r7','job':'j9','attempt':None,'event':'accepted'},{'request':'r7','job':'j9','attempt':2,'event':'worker_timeout'}]
print('One user story:',events)
Read the supplied code · 01_the_slo_you_would_actually_honour.py
"""Local mechanism demonstration for 01-the-slo-you-would-actually-honour. No AWS resources are created."""
def report(total,failures,target=.999):
if total==0: return {'state':'no eligible traffic'}
allowed=total*(1-target)
return {'success':(total-failures)/total,'allowed_failures':round(allowed,6),'budget_used':round(failures/allowed,3)}
print(report(1000000,10000)); print(report(1000,1)); print(report(0,0))
Read the supplied code · 02_the_alert_that_fires_when_it_matters_and_not_before.py
"""Local mechanism demonstration for 02-the-alert-that-fires-when-it-matters-and-not-before. No AWS resources are created."""
opened=False
for short,long in [(False,False),(True,False),(True,True),(False,True),(False,False)]:
alarm=short and long
if alarm: opened=True
print({'short':short,'long':long,'alarm':alarm,'incident_open':opened})
acknowledged=True
if acknowledged and not short and not long: opened=False
print('After acknowledged recovery:',opened)
Read the supplied code · 03_the_retry_storm_you_build_on_purpose.py
"""Local mechanism demonstration for 03-the-retry-storm-you-build-on-purpose. No AWS resources are created."""
for attempts in (3,4): print(attempts,'total attempts per layer ->',attempts**3,'dependency attempts')
remaining_ms=500; retry_after_ms=2000
print('Start another attempt:',retry_after_ms<remaining_ms)
import random
rng=random.Random(7)
print('Bounded jitter delays ms:',[round(rng.uniform(0,400)) for _ in range(5)])
Read the supplied code · 04_shedding_the_right_thing.py
"""Local mechanism demonstration for 04-shedding-the-right-thing. No AWS resources are created."""
def allocate(critical,bulk,capacity):
c=min(critical,capacity); b=min(bulk,max(0,capacity-c))
return {'critical_admitted':c,'bulk_admitted':b,'critical_rejected':critical-c,'bulk_rejected':bulk-b}
print(allocate(80,50,100)); print(allocate(120,0,100))
print('100 save-units can fund',100//100,'export or',100,'saves')
Read the supplied code · 05_the_failure_that_will_not_recover.py
"""Local mechanism demonstration for 05-the-failure-that-will-not-recover. No AWS resources are created."""
for backlog,arrivals,capacity in [(600,20,50),(600,60,50)]:
net=capacity-arrivals
print({'backlog':backlog,'net_drain_per_s':net,'ideal_drain_s':backlog/net if net>0 else 'never at these rates'})
Read the supplied code · degrade_do_not_stop.py
"""Local mechanism demonstration for degrade-do-not-stop. No AWS resources are created."""
rows=[{'owner':'ana','url':'https://example.invalid/a','title':'Old title','age':120},{'owner':'ben','url':'https://private.invalid','title':'Private','age':1}]
result=[]
for row in rows:
if row['owner']!='ana': continue
result.append({'url':row['url'],'title':row['title'] if row['age']<=3600 else None,'title_status':'stale'})
print('Provider unavailable; return local list:',result)
Read the supplied code · the_incident_you_caused_on_purpose.py
"""Local mechanism demonstration for the-incident-you-caused-on-purpose. No AWS resources are created."""
times={'inject':0,'impact':5,'alert':60,'rollback':120,'recovered':210}
print({'detection_after_impact_s':times['alert']-times['impact'],'mitigation_after_alert_s':times['rollback']-times['alert'],'recovery_after_impact_s':times['recovered']-times['impact']})
print('Rollback is a milestone; backlog clear establishes recovery.')
Read the supplied code · the_flood.py
"""Local mechanism demonstration for the-flood. No AWS resources are created."""
queued=completed=rejected=0
for second in range(10):
waiting=queued+100; done=min(20,waiting); completed+=done; waiting-=done
rejected+=max(0,waiting-300); queued=min(300,waiting)
print({'offered':1000,'completed':completed,'queued':queued,'rejected':rejected})
Read the supplied code · an_ai_feature_you_can_defend.py
"""Local mechanism demonstration for an-ai-feature-you-can-defend. No AWS resources are created."""
human=['pass']*99+['fail']; judge=['pass']*100
agreement=sum(a==b for a,b in zip(human,judge))/len(human)
failures=[i for i,x in enumerate(human) if x=='fail']
recall=sum(judge[i]=='fail' for i in failures)/len(failures)
print({'agreement':agreement,'failure_recall':recall})
Read the supplied code · the_migration_you_actually_finish.py
"""Local mechanism demonstration for the-migration-you-actually-finish. No AWS resources are created."""
target={}
def apply(version,tags,deleted=False):
if target and version<=target['version']: return 'ignored stale'
target.update(version=version,tags=tags,deleted=deleted); return 'applied'
for v,tags,deleted in [(5,['tag-db'],False),(6,[],True),(4,['old-text'],False)]:
print(v,apply(v,tags,deleted))
print('Final target:',target)
Read the supplied code · the_paved_road.py
"""Local mechanism demonstration for the-paved-road. No AWS resources are created."""
import json
def template(name,retention_days=14):
if retention_days not in (14,30): raise ValueError('choose a documented policy or record an exception')
return {'Resources':{'Logs':{'Type':'AWS::Logs::LogGroup','Properties':{'LogGroupName':'/exercise/'+name,'RetentionInDays':retention_days}}}}
print(json.dumps(template('bookmarks'),indent=2))
print('Approved alternative:',template('archive-reader',30)['Resources']['Logs']['Properties'])
actual={'RetentionInDays':None}; print('Existing resource needs review:',actual['RetentionInDays'] not in (14,30))
Read the supplied code · the_strategy_you_found_rather_than_invented.py
"""Local mechanism demonstration for the-strategy-you-found-rather-than-invented. No AWS resources are created."""
from collections import Counter
records=[('D1','transactions'),('D2','transactions'),('D3','transactions'),('D4','replay'),('D5','reporting')]
print('Constructed evidence:',records)
print('Shared requirements:',dict(Counter(kind for _,kind in records)))
print('Default: transactional store; exceptions: independently replayable events and isolated analytics.')
Read the supplied code · the_thing_you_decided_not_to_build.py
"""Local mechanism demonstration for the-thing-you-decided-not-to-build. No AWS resources are created."""
custom=6*5+12*1; script=3+12*.25; exports=2*12
print({'custom_engineer_days_year':custom,'script_engineer_days_year':script,'exports_year':exports})
print('Decision candidate: use the bounded script; revisit if demand or guarantees change.')
Read the supplied code · reading_list_it_works.py
"""Local mechanism demonstration for reading-list-it-works. No AWS resources are created."""
import sqlite3
c=sqlite3.connect(':memory:')
c.executescript('CREATE TABLE links(id INT PRIMARY KEY,owner TEXT,url TEXT,title TEXT); CREATE TABLE reads(user TEXT,link INT,is_read INT,PRIMARY KEY(user,link));')
c.execute('INSERT INTO links VALUES (?,?,?,?)',(7,'alice','https://example.invalid/guide',None))
c.executemany('INSERT INTO reads VALUES (?,?,?)',[('alice',7,1),('bob',7,0)])
print('Saved despite missing title:',c.execute('SELECT * FROM links').fetchall())
print('Personal progress:',c.execute('SELECT * FROM reads').fetchall())
Read the supplied code · reading_list_it_survives.py
"""Local mechanism demonstration for reading-list-it-survives. No AWS resources are created."""
import sqlite3
primary=sqlite3.connect(':memory:'); backup=sqlite3.connect(':memory:')
primary.execute('CREATE TABLE links(id INT PRIMARY KEY)')
primary.execute('INSERT INTO links VALUES (100)'); primary.commit(); primary.backup(backup)
primary.execute('INSERT INTO links VALUES (101)'); primary.commit()
print('Acknowledged primary:',primary.execute('SELECT id FROM links').fetchall())
print('Restored 12:00 snapshot:',backup.execute('SELECT id FROM links').fetchall())
print('Link 101 is absent: snapshot-only restore has acknowledged-write loss.')
Read the supplied code · reading_list_under_load.py
"""Local mechanism demonstration for reading-list-under-load. No AWS resources are created."""
job={'epoch':2,'state':'succeeded','title':'New title'}
def late_publish(epoch,title):
if job['epoch']!=epoch or job['state']!='running': return 'rejected stale attempt'
job['title']=title; return 'published'
print(late_publish(1,'Old title'),job)
print('Offered 200/s, origin budget 100/s: at least 100/s need cache, bounded stale data, or rejection.')
print('Ten local coalescers can still produce ten origin fills for one hot miss.')
Read the supplied code · reading_list_it_reasons.py
"""Local mechanism demonstration for reading-list-it-reasons. No AWS resources are created."""
allowed={'databases','frontend','reliability'}
proposed=['databases','finance']; confirmed=[]
invalid=set(proposed)-allowed
print('Invalid suggestions:',sorted(invalid))
print('Confirmed before user action:',confirmed)
confirmed=['databases']
print('Explicit manual confirmation:',confirmed)
Read the supplied code · reading_list_it_changes.py
"""Local mechanism demonstration for reading-list-it-changes. No AWS resources are created."""
target={'version':0,'tags':[],'deleted':False}
for version,tags,deleted in [(5,['tag-db'],False),(6,[],True),(4,['databases'],False)]:
if version>target['version']:
target.update(version=version,tags=tags,deleted=deleted); outcome='applied'
else: outcome='ignored stale'
print(version,outcome)
print('Final:',target)
AWS foundation
Use the provisioning and wiring guide for a deployable state-table, queue and private-object foundation, concrete IAM responsibilities, inspection commands and cleanup. It states exactly which application adapters remain to build.
Existing complete AI workflows
The four AI project pages retain their existing local reference implementation:
python3 examples/ai-systems/demo.py assistant
python3 examples/ai-systems/demo.py agent
python3 examples/ai-systems/demo.py extraction
python3 examples/ai-systems/demo.py evaluation
These commands use local fixtures and temporary storage. They do not invoke paid model APIs or deploy cloud resources.