LEARN. PRACTICE. GROW.Define what must survive an exit
In a fictional funds service, the contract allows CSV position exports. Before concluding that portability exists, define what the destination must reconstruct: identifiers, tenant, fund relationships, currency, monetary units, null states and read permissions. An application may communicate with an API without being able to replace its underlying service. The architecture decision should identify dependencies on identity, keys, schedules, logs and proprietary formats. Also record what the first rehearsal excludes. Our lab covers three tables and one local authorization path; it does not represent a provider migration or conversion between database engines.
Use an explicit data contract
The script exports funds, positions and grants in a read transaction with schema version, headers and file hashes. Amounts are integers in minor units and retain their currency; do not add EUR and JPY as if they were the same unit. The synthetic value 9007199254740993 was chosen to expose precision loss through float, not as a realistic customer balance. A separate flag preserves the difference between a null note and empty text. The csv module handles commas and line breaks according to the defined format. In a real migration, also add contracts for dates, time zones, decimal scales and schema evolution that this exercise does not implement.
Check relationships and atomic import
The destination enables foreign_keys before starting its transaction and confirms the setting. Relationships use identifier and tenant: A’s fund f1 does not replace B’s f1. Import loads funds first, then positions and finally permissions. If a position references a missing fund, the operation fails and the script explicitly rolls back the entire import. The exercise checks that no partial rows remain, including rows inserted before the error. Do not assume that a failed SQL statement automatically reverses every earlier statement. Beyond this example, also validate the chosen engine’s rules, indexes and behavior at larger volumes.
Reconcile meaning and access
Run python3 run.py using the complete code below. Correct import preserves rows, totals by tenant and currency, relationships and a five-request matrix. Three variants retain counts but introduce different errors: an amount loses one minor unit, a null note becomes empty and Alice’s permission moves from p17 to p18. Hashes are deliberately recalculated in these variants, demonstrating that transport integrity does not establish conversion correctness. The matrix includes allowed and denied access; merely checking that Alice can sign in would not detect every regression. Identities are supplied as trusted exercise inputs, with no authentication implementation.
Turn results into acceptance criteria
Both recorded runs passed 28 checks each on Python 3.13.1 and SQLite 3.53.4. They include rejection of altered transport data, a missing table, an unknown schema and an orphan relationship. The result evidences the executed cases; it does not establish performance, equivalent IAM policies, a production application or deletion at the previous provider. For the project, define who validates financial reconciliation, who approves permissions and who operates the destination. A hash accompanying its own file does not authenticate manifest origin. Keep the comparison reference under appropriate control and record version and scope. Connect the rehearsal with key management, recovery and provider assessment.
#!/usr/bin/env python3
"""Original BigSavant portability exercise: synthetic SQLite -> CSV -> SQLite.
No network, cloud APIs, credentials or external packages. Not a migration tool.
"""
import copy
import csv
import datetime
import hashlib
import io
import json
from pathlib import Path
import platform
import sqlite3
checks = []
def check(name, inputs, actual, expected):
if actual!= expected:
raise AssertionError(f"{name}: {actual!r}!= {expected!r}")
checks.append(dict(name=name, inputs=inputs, actual=actual, expected=expected))
SCHEMA = """
CREATE TABLE funds(id TEXT NOT NULL, tenant TEXT NOT NULL, name TEXT NOT NULL,
PRIMARY KEY(id,tenant));
CREATE TABLE positions(id TEXT NOT NULL, tenant TEXT NOT NULL, fund_id TEXT NOT NULL,
currency TEXT NOT NULL, amount_minor INTEGER NOT NULL, note TEXT,
PRIMARY KEY(id,tenant), FOREIGN KEY(fund_id,tenant) REFERENCES funds(id,tenant));
CREATE TABLE grants(subject TEXT NOT NULL, tenant TEXT NOT NULL, position_id TEXT NOT NULL,
action TEXT NOT NULL CHECK(action IN ('read','write')),
PRIMARY KEY(subject,tenant,position_id,action),
FOREIGN KEY(position_id,tenant) REFERENCES positions(id,tenant));
"""
COLUMNS = {
'funds': ['id','tenant','name'],
'positions': ['id','tenant','fund_id','currency','amount_minor','note','note_is_null'],
'grants': ['subject','tenant','position_id','action']
}
ORDER = {'funds':'tenant,id', 'positions':'tenant,id', 'grants':'tenant,subject,position_id,action'}
def database:
db = sqlite3.connect(':memory:', isolation_level=None)
db.execute('PRAGMA foreign_keys=ON')
if db.execute('PRAGMA foreign_keys').fetchone[0]!= 1:
raise RuntimeError('Foreign keys unavailable')
db.executescript(SCHEMA)
return db
def snapshot(db):
return {t:db.execute(f'SELECT * FROM {t} ORDER BY {ORDER[t]}').fetchall for t in COLUMNS}
def digest(data):
return hashlib.sha256(data).hexdigest
def csv_bytes(table, rows):
out = io.StringIO(newline='')
writer = csv.writer(out, lineterminator='\n')
writer.writerow(COLUMNS[table])
for row in rows:
values = list(row)
if table == 'positions':
values.append('1' if values[5] is None else '0')
writer.writerow(values)
return out.getvalue.encode('utf-8')
def export(db):
db.execute('BEGIN')
try:
state = snapshot(db)
bundle = {'schemaVersion':1, 'files':{}, 'manifest':{}}
for t, rows in state.items:
data = csv_bytes(t, rows)
bundle['files'][t] = data
bundle['manifest'][t] = {'rows':len(rows),'sha256':digest(data)}
db.execute('COMMIT')
return bundle
except Exception:
db.execute('ROLLBACK')
raise
def load(bundle, db):
if bundle['schemaVersion']!= 1 or set(bundle['files'])!= set(COLUMNS):
raise ValueError('Unsupported schema or missing table')
# Hashes check transport consistency, not authenticity of the manifest.
for t, data in bundle['files'].items:
if digest(data)!= bundle['manifest'][t]['sha256']:
raise ValueError('Transport digest mismatch')
db.execute('BEGIN')
try:
for t in COLUMNS: # parents, children, then grants
reader = csv.reader(io.StringIO(bundle['files'][t].decode('utf-8'), newline=''))
if next(reader)!= COLUMNS[t]:
raise ValueError('Column mismatch')
rows = list(reader)
if len(rows)!= bundle['manifest'][t]['rows']:
raise ValueError('Row count mismatch')
for row in rows:
if len(row)!= len(COLUMNS[t]):
raise ValueError('Column count mismatch')
if t == 'positions':
if row[6] not in ('0','1') or (row[6]=='1' and row[5]!=''):
raise ValueError('Null flag mismatch')
amount = int(row[4])
if str(amount)!= row[4]:
raise ValueError('Noncanonical integer')
row = row[:4]+[amount, None if row[6]=='1' else row[5]]
db.execute(f"INSERT INTO {t} VALUES ({','.join('?' for _ in row)})", row)
if db.execute('PRAGMA foreign_key_check').fetchall:
raise ValueError('Foreign key check failed')
db.execute('COMMIT')
except Exception:
db.execute('ROLLBACK')
raise
def allowed(db, subject, trusted_tenant, position_id, object_tenant, action):
# Identity and tenant are supplied as trusted inputs, not authenticated here.
if trusted_tenant!= object_tenant:
return False
return db.execute('''SELECT 1 FROM positions p JOIN grants g
ON g.position_id=p.id AND g.tenant=p.tenant
WHERE p.id=? AND p.tenant=? AND g.subject=? AND g.action=?''',
(position_id, object_tenant, subject, action)).fetchone is not None
REQUESTS = [
('alice','A','p17','A','read'),
('alice','A','p18','A','read'),
('alice','A','p17','B','read'),
('alice','A','p17','A','write'),
('bob','B','p17','B','read')
]
def totals(db):
return db.execute('''SELECT tenant,currency,SUM(amount_minor) FROM positions
GROUP BY tenant,currency ORDER BY tenant,currency''').fetchall
def assess(src, dst):
a,b = snapshot(src),snapshot(dst)
return {
'rowCountsMatch':{t:len(v) for t,v in a.items}=={t:len(v) for t,v in b.items},
'canonicalRowsMatch':a==b,
'amountsByTenantCurrencyMatch':totals(src)==totals(dst),
'authorizationMatrixMatches':[allowed(src,*q) for q in REQUESTS]==[allowed(dst,*q) for q in REQUESTS],
'foreignKeysValid':not dst.execute('PRAGMA foreign_key_check').fetchall
}
def mutate_csv(bundle, table, transform, refresh_digest=True):
b = copy.deepcopy(bundle)
rows = list(csv.reader(io.StringIO(b['files'][table].decode('utf-8'), newline='')))
transform(rows)
out = io.StringIO(newline=''); csv.writer(out,lineterminator='\n').writerows(rows)
b['files'][table] = out.getvalue.encode('utf-8')
if refresh_digest:
b['manifest'][table] = {'rows':len(rows)-1,'sha256':digest(b['files'][table])}
return b
src = database
src.executemany('INSERT INTO funds VALUES(?,?,?)',[('f1','A','Fundo A, Lisboa'),('f1','B','Fund B')])
src.executemany('INSERT INTO positions VALUES(?,?,?,?,?,?)',[
('p17','A','f1','EUR',9007199254740993,None),
('p18','A','f1','EUR',700000001,''),
('p17','B','f1','JPY',2345,'Liquidação, Lisboa\nLine 2')
])
src.executemany('INSERT INTO grants VALUES(?,?,?,?)',[('alice','A','p17','read'),('bob','B','p17','read')])
bundle = export(src)
dst = database; load(bundle,dst)
baseline = assess(src,dst)
check('complete export restores all defined invariants',{'schemaVersion':1,'tables':list(COLUMNS)},baseline,dict.fromkeys(baseline,True))
check('explicit foreign keys enabled',{'database':'restored'},dst.execute('PRAGMA foreign_keys').fetchone[0],1)
check('integer above binary64 exact range preserved',{'column':'amount_minor'},str(dst.execute("SELECT amount_minor FROM positions WHERE tenant='A' AND id='p17'").fetchone[0]),'9007199254740993')
check('NULL and empty text remain distinct',{'tenant':'A'},[x[0] for x in dst.execute("SELECT note FROM positions WHERE tenant='A' ORDER BY id")],[None,''])
check('comma newline and Unicode survive CSV',{'tenant':'B'},dst.execute("SELECT note FROM positions WHERE tenant='B'").fetchone[0],'Liquidação, Lisboa\nLine 2')
check('authorization matrix preserved',{'requests':[list(q) for q in REQUESTS]},[allowed(dst,*q) for q in REQUESTS],[True,False,False,False,True])
# Valid transport hashes do not prove that a conversion preserved meaning.
rounded = mutate_csv(bundle,'positions',lambda rows:rows[1].__setitem__(4,str(int(float(rows[1][4])))))
lossy = database; load(rounded,lossy); verdict = assess(src,lossy)
check('rounded import retains row counts',{'mutation':'binary64 conversion'},verdict['rowCountsMatch'],True)
check('rounded import changes one minor unit',{'mutation':'binary64 conversion'},str(lossy.execute("SELECT amount_minor FROM positions WHERE tenant='A' AND id='p17'").fetchone[0]),'9007199254740992')
check('amount reconciliation detects rounding',{'mutation':'binary64 conversion'},verdict['amountsByTenantCurrencyMatch'],False)
check('full row comparison detects rounding',{'mutation':'binary64 conversion'},verdict['canonicalRowsMatch'],False)
collapsed = mutate_csv(bundle,'positions',lambda rows:rows[1].__setitem__(6,'0'))
null_loss = database; load(collapsed,null_loss); verdict = assess(src,null_loss)
check('NULL collapse retains counts and totals',{'mutation':'NULL becomes empty'},[verdict['rowCountsMatch'],verdict['amountsByTenantCurrencyMatch']],[True,True])
check('full row comparison detects NULL collapse',{'mutation':'NULL becomes empty'},verdict['canonicalRowsMatch'],False)
grant_change = mutate_csv(bundle,'grants',lambda rows:rows[1].__setitem__(2,'p18'))
wrong_grants = database; load(grant_change,wrong_grants); verdict = assess(src,wrong_grants)
check('changed grants retain counts and amounts',{'mutation':'alice grant moved p17 to p18'},[verdict['rowCountsMatch'],verdict['amountsByTenantCurrencyMatch']],[True,True])
check('negative access test detects new permission',{'request':list(REQUESTS[1])},allowed(wrong_grants,*REQUESTS[1]),True)
check('authorization matrix detects grant regression',{'mutation':'alice grant moved p17 to p18'},verdict['authorizationMatrixMatches'],False)
for name,bad in [
('orphan relationship',mutate_csv(bundle,'positions',lambda rows:rows[2].__setitem__(2,'missing-fund'))),
('transport corruption',mutate_csv(bundle,'positions',lambda rows:rows[2].__setitem__(4,'1'),False)),
('missing grants file',{**bundle,'files':{k:v for k,v in bundle['files'].items if k!='grants'}}),
('unsupported schema',{**bundle,'schemaVersion':2})
]:
candidate = database
try:
load(bad,candidate); rejected=False
except (ValueError,sqlite3.IntegrityError):
rejected=True
check(name+' rejected',{'fault':name},rejected,True)
check(name+' leaves no partial import',{'fault':name},sum(len(v) for v in snapshot(candidate).values),0)
candidate.close
# Demonstrate why a source snapshot is stale after new destination writes.
dst.execute("UPDATE positions SET amount_minor=amount_minor+7 WHERE tenant='B' AND id='p17'")
check('destination write diverges from source snapshot',{'newDestinationMinorUnits':7},assess(src,dst)['amountsByTenantCurrencyMatch'],False)
check('source snapshot omits post-cutover write',{'tenant':'B'},src.execute("SELECT amount_minor FROM positions WHERE tenant='B'").fetchone[0],2345)
check('destination contains post-cutover write',{'tenant':'B'},dst.execute("SELECT amount_minor FROM positions WHERE tenant='B'").fetchone[0],2352)
# This is an original planning calculation, not observed elapsed recovery time.
window, decision, restore, validation = 120,80,25,20
check('late rollback decision exceeds window',{'windowMinutes':window,'decisionMinute':decision,'restoreMinutes':restore,'validationMinutes':validation},decision+restore+validation>window,True)
check('latest calculated rollback decision',{'windowMinutes':window,'restoreMinutes':restore,'validationMinutes':validation},window-restore-validation,75)
for db in (src,dst,lossy,null_loss,wrong_grants):
db.close
print(json.dumps(dict(checkedAt=datetime.datetime.now(datetime.timezone.utc).isoformat,python=platform.python_version,sqlite=sqlite3.sqlite_version,scriptSha256=digest(Path(__file__).read_bytes),network=False,groups=['SQLite/CSV semantic restoration','Authorization preservation','Failed-import atomicity','Post-cutover divergence and planning'],checks=checks,limitations=[
'Actual in-memory SQLite and CSV operations; no cloud-provider migration, database-engine conversion, production application, IAM or network execution.',
'Synthetic data only. Authorization uses supplied trusted identities and one local SQL path; authentication and provider policy translation are not implemented.',
'Transport hashes are not signed and do not authenticate the manifest. Mutated bundles deliberately recompute hashes to demonstrate semantic failures.',
'No concurrent writes, load test, measured RTO/RPO, timed cutover or automatic reverse replication. Rollback timing is an explicit planning calculation.',
'No physical deletion, legal compliance attestation, contractual enforcement or independent audit demonstrated.'
]),ensure_ascii=False,indent=2))
Correct position counts can coexist with incorrect amounts or permissions.
Common pitfalls
Confusing export with portability; hashes with functional equivalence; old infrastructure with safe rollback; estimates with measured timings.
Related topics: Key management and restoration · Risk, suppliers and continuity
Execute export and restoration, detecting semantic loss and access regressions.
Reference: SQLite foreign key support · CCSP examination outline effective 2026-08-01; January2026 V2 PDF