Master/data-meta-model
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
data-meta-model/scripts/migrate_instances_v1_2.py
T

208 lines
8.5 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""
SODH - migration v1.1 -> v1.2
==============================
USAGE
python3 scripts/migrate_instances_v1_2.py # dry run
python3 scripts/migrate_instances_v1_2.py --apply # rewrite, .bak kept
WHAT IT DOES
1. physicalizedIn -> storedIn everywhere (no instances yet, done for
completeness so the term never appears again).
2. hasSource removed from every Business Concept. A concept has no source
of truth: its authority is its owning domain, its definition belongs in
the glossary. Asking which system defines it has no answer, because
every system that uses it would be a candidate.
3. hasSource on Metrics recomputed from the elements that implement them.
It said "SODH back-doc v0.7" -- the provenance of the record, not the
source of the data. Asserted rather than derived, by exception, because
"where can I get this metric" is a first-order question and the
derivation chain will have gaps until harvesting; a shape then checks
the assertion against the elements.
4. The SODH Data Product, its contract and its two interfaces.
WHY THE PRODUCT OWNER IS EMPTY
Nobody is appointed. The field stays absent and the shape reports a
warning rather than a violation. Inventing a name to satisfy a blocking
rule is exactly how BR-004 produced padding in v0.6.
"""
import os
import re
import shutil
import sys
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
TTL = os.path.join(REPO, "instances", "sodh.ttl")
W = 78
PRODUCT = "ex:DP_06_01_001"
CONTRACT = "ex:DC_06_01_001"
IFACE_WH = "ex:DI_06_01_001"
IFACE_FILE = "ex:DI_06_01_002"
def clauses(body):
out, buf, i, n = [], [], 0, len(body)
while i < n:
c = body[i]
if c == '"':
j = i + 1
while j < n and body[j] != '"':
j += 2 if body[j] == "\\" else 1
j = min(j + 1, n)
buf.append(body[i:j]); i = j; continue
if c == ";":
out.append("".join(buf)); buf = []; i += 1; continue
buf.append(c); i += 1
out.append("".join(buf))
return out
def set_prop(text, subject, prop, value):
pat = re.compile(r'(^%s a pr:\w+ ;)(.*?)(\.\s*\n)' % re.escape(subject), re.S | re.M)
m = pat.search(text)
if not m:
return text, False
kept = [c for c in clauses(m.group(2))
if c.strip() and not re.match(r'\s*%s\s' % re.escape(prop), c)]
if value is not None:
kept.insert(0, "\n %s %s " % (prop, value))
body = " ;".join(kept).rstrip() + "\n "
return text[:m.start()] + m.group(1) + body + m.group(3) + text[m.end():], True
def get_prop(text, subject, prop):
m = re.search(r'^%s a pr:\w+ ;(.*?)\.\s*\n' % re.escape(subject), text, re.S | re.M)
if not m:
return None
for c in clauses(m.group(1)):
mm = re.match(r'\s*%s\s+(.*)$' % re.escape(prop), c, re.S)
if mm:
return mm.group(1).strip()
return None
def product_blocks(objects):
packages = " ,\n ".join(objects)
return """# --- data product ------------------------------------------------------
# The unit of delivery: what carries a version, an owner, a changelog and a
# list of consumers to warn. It has no single technical counterpart on purpose
# -- rename the schema and the product is unchanged; change its SLA and it is
# not. Its physical footprint is derivable through packages, never asserted.
%s a pr:DataProduct ;
pr:hasIdentifier "DP-06.01-001" ; pr:hasName "Sell Out Data Hub - Gold" ;
pr:owningDomain ex:DD_06 ;
pr:operatedBy ex:SD_06_01 ;
pr:packages %s ;
pr:governedBy %s ;
pr:exposes %s , %s ;
pr:hasStatus "DRAFT" ; pr:hasVersion "1.2" ; pr:hasSource "SODH Gold" .
%s a pr:DataContract ;
pr:hasIdentifier "DC-06.01-001" ; pr:hasName "Sell Out Data Hub - Gold service commitment" ;
pr:owningDomain ex:DD_06 ;
pr:hasBusinessRule "Refreshed weekly, available Tuesday 06:00 CET. Panel coverage published with every load. Schema stable within a minor version; 60 days notice on any breaking change. Support through the Sales Performance data team." ;
pr:hasStatus "DRAFT" ; pr:hasVersion "1.2" .
%s a pr:DataInterface ;
pr:hasIdentifier "DI-06.01-001" ; pr:hasName "SODH Gold consumption schema" ;
pr:owningDomain ex:DD_06 ;
pr:hasBusinessDefinition "Read access to the Gold layer through Snowflake grants. The channel BI tools and analysts consume."@en ;
pr:hasStatus "DRAFT" ; pr:hasVersion "1.2" ; pr:hasSource "SODH Gold" .
%s a pr:DataInterface ;
pr:hasIdentifier "DI-06.01-002" ; pr:hasName "Panel partner weekly extract" ;
pr:owningDomain ex:DD_06 ;
pr:hasBusinessDefinition "File extract delivered to the panel provider each week. Outside Snowflake, and the boundary where lineage leaves the warehouse."@en ;
pr:hasStatus "DRAFT" ; pr:hasVersion "1.2" ; pr:hasSource "SODH Gold" .
""" % (PRODUCT, packages, CONTRACT, IFACE_WH, IFACE_FILE,
CONTRACT, IFACE_WH, IFACE_FILE)
def main():
apply_changes = "--apply" in sys.argv
if not os.path.exists(TTL):
print("Not found: %s" % TTL); sys.exit(2)
text = original = open(TTL, encoding="utf-8").read()
report = []
# 1. term rename
text, n = re.subn(r'\bpr:physicalizedIn\b', 'pr:storedIn', text)
report.append("physicalizedIn -> storedIn: %d occurrence(s)" % n)
# 2. no source of truth on a concept
n_bc = 0
for m in list(re.finditer(r'^(ex:BC_\w+) a pr:BusinessConcept ;', text, re.M)):
if get_prop(text, m.group(1), "pr:hasSource") is not None:
text, _ = set_prop(text, m.group(1), "pr:hasSource", None)
n_bc += 1
report.append("hasSource removed from %d business concepts" % n_bc)
# 3. metric source of truth, from the elements that implement it
de_src = {}
for m in re.finditer(r'^(ex:DE_\w+) a pr:DataElement ;(.*?)\.\s*\n', text, re.S | re.M):
v = re.search(r'pr:hasSource\s+"([^"]*)"', m.group(2))
if v:
de_src[m.group(1)] = v.group(1)
wired = {}
for m in re.finditer(r'(ex:M_\w+)\s+pr:computedBy(.*?)\.', text, re.S):
wired[m.group(1)] = re.findall(r'ex:DE_\w+', m.group(2))
n_m, mixed = 0, []
for metric, des in wired.items():
srcs = sorted({de_src[d] for d in des if d in de_src})
if not srcs:
continue
if len(srcs) > 1:
mixed.append(metric)
text, ok = set_prop(text, metric, "pr:hasSource",
" , ".join('"%s"' % x for x in srcs))
n_m += ok
report.append("source of truth recomputed on %d metrics%s"
% (n_m, "" if not mixed else
" (%d blend several systems)" % len(mixed)))
# 4. the data product
if not re.search(r'^%s a ' % re.escape(PRODUCT), text, re.M):
objs = re.findall(r'^(ex:DO_\w+) a pr:DataObject ;', text, re.M)
text = text.rstrip() + "\n\n\n" + product_blocks(objs)
report.append("data product created, packaging %d data objects" % len(objs))
else:
report.append("data product already present, skipped")
# ---- report
print()
print("SODH MIGRATION v1.1 -> v1.2 %s" % ("APPLY" if apply_changes else "DRY RUN"))
print("=" * W)
for line in report:
print(" " + line)
print("=" * W)
bc_src = len(re.findall(r'^ex:BC_\w+ a pr:BusinessConcept ;(?:(?!\.\s*\n).)*?hasSource',
text, re.S | re.M))
defined = set(re.findall(r'^(ex:\w+) a ', text, re.M))
dangling = sorted(set(re.findall(r'\bex:\w+', text)) - defined)
doubled = []
for m in re.finditer(r'^(ex:\w+) a pr:\w+ ;(.*?)\.\s*\n', text, re.S | re.M):
for p in ("pr:hasName", "pr:hasSource", "pr:operatedBy", "pr:governedBy"):
if len(re.findall(r'(?<![\w:])%s\s' % re.escape(p), m.group(2))) > 1:
doubled.append("%s/%s" % (m.group(1), p))
print(" concepts still carrying a source: %d (must be 0)" % bc_src)
print(" physicalizedIn left: %d (must be 0)"
% len(re.findall(r'pr:physicalizedIn', text)))
print(" dangling references: %s" % (", ".join(dangling) if dangling else "none"))
print(" duplicated single-valued properties: %s"
% (", ".join(doubled) if doubled else "none"))
print("=" * W)
if text != original and apply_changes:
shutil.copy2(TTL, TTL + ".bak")
open(TTL, "w", encoding="utf-8").write(text)
print(" written, backup at %s.bak" % os.path.basename(TTL))
elif text != original:
print(" dry run -- re-run with --apply to write")
print()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink