Recover reports, models and sharing

"""Original offline exercise: reconcile join cardinality with fixed in-memory SQL."""
import json
import re
import sqlite3
import sys


def exact_keys(row, keys):
 if not isinstance(row, dict) or set(row)!= set(keys):
 raise ValueError('Unexpected object fields')


def integer(value, low, high):
 if type(value) is not int or not low <= value <= high:
 raise ValueError('Integer outside contract')


def token(value, currency=False):
 pattern = r'[A-Z]{3}' if currency else r'[A-Za-z0-9_-]{1,48}'
 if not isinstance(value, str) or re.fullmatch(pattern, value) is None:
 raise ValueError('Invalid identifier')


def reconcile(data):
 exact_keys(data, ['trades', 'rates'])
 for key in ['trades', 'rates']:
 if not isinstance(data[key], list) or len(data[key]) > 100:
 raise ValueError('Expected at most100rows per collection')
 seen = set
 for trade in data['trades']:
 exact_keys(trade, ['id', 'book', 'currency', 'amountMinor'])
 token(trade['id']); token(trade['book']); token(trade['currency'], True)
 integer(trade['amountMinor'], -1000000, 1000000)
 if trade['id'] in seen:
 raise ValueError('Duplicate trade identifier')
 seen.add(trade['id'])
 for rate in data['rates']:
 exact_keys(rate, ['currency', 'partsPerMillion'])
 token(rate['currency'], True)
 integer(rate['partsPerMillion'], 1, 100000000)
 # Even100x100matches with products of10^14 stay below signed64-bit SUM.
 db = sqlite3.connect(':memory:')
 db.row_factory = sqlite3.Row
 try:
 db.executescript('''
 CREATE TABLE trades(id TEXT PRIMARY KEY,book TEXT NOT NULL,
 currency TEXT NOT NULL,amount INTEGER NOT NULL);
 CREATE TABLE rates(currency TEXT NOT NULL,ppm INTEGER NOT NULL);
 ''')
 db.executemany('INSERT INTO trades VALUES(?,?,?,?)',
 [(t['id'],t['book'],t['currency'],t['amountMinor']) for t in data['trades']])
 db.executemany('INSERT INTO rates VALUES(?,?)',
 [(r['currency'],r['partsPerMillion']) for r in data['rates']])
 diagnostics = [dict(r) for r in db.execute('''
 SELECT t.id,COUNT(*) AS leftRows,COUNT(r.currency) AS matches,
 SUM(t.amount*r.ppm) AS scaledSum
 FROM trades t LEFT JOIN rates r ON t.currency=r.currency
 GROUP BY t.id ORDER BY t.id
 ''')]
 preview = [dict(r) for r in db.execute('''
 SELECT t.book,COUNT(*) AS joinedRows,SUM(t.amount*r.ppm) AS scaledSum
 FROM trades t INNER JOIN rates r ON t.currency=r.currency
 GROUP BY t.book ORDER BY t.book
 ''')]
 missing = [r['id'] for r in diagnostics if r['matches'] == 0]
 duplicated = [r['id'] for r in diagnostics if r['matches'] > 1]
 passed = not missing and not duplicated
 return dict(sqliteVersion=sqlite3.sqlite_version,inputTrades=len(data['trades']),
 joinedRows=sum(r['matches'] for r in diagnostics),
 diagnostics=diagnostics,missingTradeIds=missing,
 multipliedTradeIds=duplicated,unsafePreview=preview,
 cardinalityPassed=passed,reconciledReport=preview if passed else None,
 scaleDenominator=1000000,stateLifetime='one invocation; memory only',
 sourceCompletenessProven=False,rateDateValidityProven=False,
 cloudSemanticsValidated=False,productionPublicationApproved=False)
 finally:
 db.close


def unique_object(pairs):
 row = {}
 for key, value in pairs:
 if key in row:
 raise ValueError('Duplicate JSON key')
 row[key] = value
 return row


if __name__ == '__main__':
 try:
 raw = sys.stdin.buffer.read(2000001)
 if len(raw) > 2000000:
 raise ValueError('Input exceeds2MB')
 result = reconcile(json.loads(raw, object_pairs_hook=unique_object))
 print(json.dumps(result, ensure_ascii=False, sort_keys=True))
 except (ValueError, TypeError, UnicodeError, RecursionError) as exc:
 print(json.dumps({'error': str(exc)}), file=sys.stderr)
 sys.exit(2)

Related topics: Cardinality and reconciliation · BigQuery ML and labels · Authorized views and subscriptions

Reference: Professional Data Engineer standard exam guide · Current linked standard guide (document title v4.2); edition date unconfirmed (2026-09-30 inspection)