Tryton/tryton-db-migration
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rewritten for readability by Claude

Browse Source
This commit is contained in:
uha4
2026-08-09 14:13:21 +02:00
parent 388ee649c3
commit 5f11c77ddf
1 changed files with 297 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extract.py
+297 -199
View File
@@ -1,4 +1,9 @@
#!/usr/bin/env python3
"""
Analysiert __register__-Methoden in Tryton-Modulen und extrahiert alle
Datenbank-Operationen (Spalten-/Tabellen-Änderungen, execute/insert/update/...),
die dort ausgeführt werden. Ergebnis wird pro Modul als JSON abgelegt.
"""
import re
import ast
@@ -6,6 +11,11 @@ import argparse
import json
from pathlib import Path
# --------------------------------------------------------------------------
# Konfiguration / Defaults
# --------------------------------------------------------------------------
DEFAULT_REPO = "/home/uha4/tmp/tryton"
DEFAULT_MODULES = [
@@ -27,18 +37,84 @@ DEFAULT_MODULES = [
DEFAULT_OUTPUT = "data"
# Operationen, die find_tables() als "hier wird eine Tabelle referenziert"
# erkennt (inkl. table_exist, das selbst keine Änderung ist, aber die
# betroffene Tabelle offenlegt).
TABLE_REFERENCE_OPS = {
"column_rename",
"drop_column",
"add_column",
"drop_constraint",
"add_constraint",
"not_null_action",
"index_action",
"alter_size",
"update",
"insert",
"delete",
"select",
"table_exist",
}
# Operationen, die extract_call() tatsächlich als DB-Operation aufnimmt.
# execute/executemany kommen hier zusätzlich dazu, table_exist bewusst nicht
# (das ist nur eine Existenzprüfung, keine Änderung).
INTERESTING_OPS = {
"column_rename",
"drop_column",
"add_column",
"drop_constraint",
"add_constraint",
"not_null_action",
"index_action",
"alter_size",
"execute",
"executemany",
"update",
"insert",
"delete",
"select",
}
# --------------------------------------------------------------------------
# AST-Visitor
# --------------------------------------------------------------------------
class RegisterExtractor(ast.NodeVisitor):
"""
Läuft über eine Python-Datei und sammelt für jede __register__-Methode
die darin ausgeführten DB-Operationen (vor/nach dem super()-Aufruf,
inkl. eventueller if-Bedingung).
"""
def __init__(self, source, module, filename):
self.module = module
self.filename = filename
self.source = source
self.results = []
# Name der Model-Klasse -> __name__ (z. B. "MyClass" -> "ir.model.data")
self.model_names = {}
# Innerhalb der aktuell analysierten __register__-Methode:
# lokale Variable -> Tabellenname
self.table_vars = {}
# lokale Variable -> Quelltext-Ausdruck, aus dem sie stammt
# (für die spätere Ersetzung im "code"-Feld der Operation)
self.alias_vars = {}
# Klasse, in der wir uns gerade befinden (für cls._table etc.)
self.current_class = None
# -- Modul-/Klassenebene ------------------------------------------------
def visit_Module(self, node):
# Erst alle __name__-Zuweisungen einsammeln, damit sie beim
# eigentlichen Durchlauf schon bekannt sind (Reihenfolge der
# Klassen im File soll keine Rolle spielen).
for item in node.body:
if isinstance(item, ast.ClassDef):
model_name = self.get_model_name(item)
@@ -54,20 +130,41 @@ class RegisterExtractor(ast.NodeVisitor):
self.generic_visit(node)
def extract_register(self, cls, func):
def get_model_name(self, node):
"""Liest `__name__ = "some.model"` aus einem Klassenkörper."""
for item in node.body:
if not isinstance(item, ast.Assign):
continue
for target in item.targets:
if isinstance(target, ast.Name) and target.id == "__name__":
try:
return ast.literal_eval(item.value)
except Exception:
return None
return None
# -- __register__ selbst -------------------------------------------------
def extract_register(self, cls, func):
"""Analysiert eine einzelne __register__-Methode."""
operations = []
self.table_vars = {}
self.alias_vars = {}
self.current_class = cls
model_name = self.model_names.get(cls)
if model_name:
self.table_vars["cls"] = model_name
# Tabellen-Variablen (z. B. `table = cls.__table__()`) vorab
# einsammeln, damit sie beim eigentlichen Walk schon aufgelöst
# werden können.
for stmt in func.body:
self.collect_table_assignment(stmt)
self.walk_statements(func.body, operations, "before_super", None)
self.walk_statements(func.body, operations, phase="before_super", condition=None)
self.results.append({
"module": self.module,
@@ -78,25 +175,14 @@ class RegisterExtractor(ast.NodeVisitor):
"operations": operations,
})
def get_model_name(self, node):
for item in node.body:
if not isinstance(item, ast.Assign):
continue
for target in item.targets:
if (
isinstance(target, ast.Name)
and target.id == "__name__"
):
try:
return ast.literal_eval(item.value)
except Exception:
return None
return None
# -- Tabellen-Variablen auflösen -----------------------------------------
def collect_table_assignment(self, stmt):
"""
Sucht rekursiv nach Zuweisungen der Form `name = <table-ausdruck>`
und merkt sich in self.table_vars, welche Variable welche Tabelle
referenziert.
"""
for node in ast.walk(stmt):
if not isinstance(node, ast.Assign):
continue
@@ -129,22 +215,19 @@ class RegisterExtractor(ast.NodeVisitor):
self.table_vars[target.id] = node.value.value
def resolve_table_call(self, node):
"""
Erkennt Aufrufe, die direkt eine Tabelle liefern:
- pool.get('some.model')
- X.__table__() / X.__table_handler__()
und gibt den zugehörigen (normalisierten) Tabellennamen zurück.
"""
if not isinstance(node, ast.Call):
return None
# ModelClass = pool.get('ir.model.data')
if (
isinstance(node.func, ast.Attribute)
and node.func.attr == "get"
and isinstance(node.func.value, ast.Name)
and node.func.value.id == "pool"
and len(node.args) == 1
and isinstance(node.args[0], ast.Constant)
and isinstance(node.args[0].value, str)
):
if self._is_pool_get(node):
return node.args[0].value
if not (
isinstance(node.func, ast.Attribute)
and node.func.attr in ("__table__", "__table_handler__")
@@ -152,23 +235,11 @@ class RegisterExtractor(ast.NodeVisitor):
return None
receiver = node.func.value
# ModelClass = pool.get('ir.model.data')
if (
isinstance(receiver, ast.Name)
and receiver.id == "pool"
):
# pool.__table__() o.ä. ergibt keinen sinnvollen Tabellennamen
if isinstance(receiver, ast.Name) and receiver.id == "pool":
return None
if (
isinstance(node.func, ast.Attribute)
and node.func.attr == "get"
and isinstance(node.func.value, ast.Name)
and node.func.value.id == "pool"
and len(node.args) == 1
and isinstance(node.args[0], ast.Constant)
and isinstance(node.args[0].value, str)
):
return node.args[0].value
if isinstance(receiver, ast.Name):
if receiver.id == "cls":
model_name = self.model_names.get(self.current_class)
@@ -177,16 +248,33 @@ class RegisterExtractor(ast.NodeVisitor):
self.model_names.get(receiver.id)
or self.table_vars.get(receiver.id)
)
if model_name:
return model_name.replace(".", "_")
return None
@staticmethod
def _is_pool_get(node):
"""True für Aufrufe der Form `pool.get('some.model')`."""
return (
isinstance(node.func, ast.Attribute)
and node.func.attr == "get"
and isinstance(node.func.value, ast.Name)
and node.func.value.id == "pool"
and len(node.args) == 1
and isinstance(node.args[0], ast.Constant)
and isinstance(node.args[0].value, str)
)
def resolve_table_object(self, node):
"""
Löst einen beliebigen Ausdruck (Variable, cls._table, Aufruf) zu
einem Tabellennamen auf, falls möglich.
"""
if isinstance(node, ast.Name):
return self.table_vars.get(node.id)
if (
isinstance(node, ast.Attribute)
and node.attr == "_table"
@@ -196,10 +284,14 @@ class RegisterExtractor(ast.NodeVisitor):
model_name = self.model_names.get(self.current_class)
if model_name:
return model_name.replace(".", "_")
return self.resolve_table_call(node)
def find_tables(self, node):
"""
Sammelt alle Tabellen, die innerhalb eines Ausdrucks (typischerweise
eines Funktionsaufrufs) referenziert werden.
"""
tables = []
for child in ast.walk(node):
@@ -208,46 +300,45 @@ class RegisterExtractor(ast.NodeVisitor):
if not (
isinstance(child.func, ast.Attribute)
and child.func.attr in {
"column_rename",
"drop_column",
"add_column",
"drop_constraint",
"add_constraint",
"not_null_action",
"index_action",
"alter_size",
"update",
"insert",
"delete",
"select",
"table_exist",
}
and child.func.attr in TABLE_REFERENCE_OPS
):
continue
# backend.TableHandler.table_exist(cls._table)
if (
isinstance(child.func.value, ast.Attribute)
and child.func.value.attr == "TableHandler"
and isinstance(child.func.value.value, ast.Name)
and child.func.value.value.id == "backend"
and child.func.attr == "table_exist"
and child.args
):
if self._is_table_handler_call(child, attr="table_exist") and child.args:
table = self.resolve_table_object(child.args[0])
if table and table not in tables:
tables.append(table)
continue
table = self.resolve_table_object(child.func.value)
if table and table not in tables:
tables.append(table)
return tables
def walk_statements(self, statements, operations, phase, condition):
@staticmethod
def _is_table_handler_call(call, attr=None):
"""True für Aufrufe der Form `backend.TableHandler.<attr>(...)`."""
is_handler = (
isinstance(call.func.value, ast.Attribute)
and call.func.value.attr == "TableHandler"
and isinstance(call.func.value.value, ast.Name)
and call.func.value.value.id == "backend"
)
if not is_handler:
return False
return attr is None or call.func.attr == attr
# -- Statements/Ausdrücke durchlaufen ------------------------------------
def walk_statements(self, statements, operations, phase, condition):
"""
Läuft über eine Liste von Statements und sammelt DB-Operationen ein.
`phase` wechselt von "before_super" auf "after_super", sobald der
super().__register__(...)-Aufruf gefunden wird. `condition` trägt
den Text einer umschließenden if-Bedingung mit, falls vorhanden.
"""
for stmt in statements:
if self.is_super_register(stmt):
@@ -265,7 +356,6 @@ class RegisterExtractor(ast.NodeVisitor):
self.walk_expression(value, operations, phase, condition)
def walk_expression(self, node, operations, phase, condition):
if node is None:
return
@@ -281,21 +371,16 @@ class RegisterExtractor(ast.NodeVisitor):
self.walk_expression(child, operations, phase, condition)
def is_super_register(self, stmt):
"""True für `super(...).__register__(...)` als eigenständiges Statement."""
if not isinstance(stmt, ast.Expr):
return False
call = stmt.value
if not isinstance(call, ast.Call):
return False
func = call.func
if not isinstance(func, ast.Attribute):
return False
if func.attr != "__register__":
if not isinstance(func, ast.Attribute) or func.attr != "__register__":
return False
if not isinstance(func.value, ast.Call):
@@ -306,99 +391,44 @@ class RegisterExtractor(ast.NodeVisitor):
return func.value.func.id == "super"
# -- Einzelnen Aufruf in ein Operation-Dict übersetzen -------------------
def extract_call(self, call):
if not isinstance(call.func, ast.Attribute):
return None
interesting = {
"column_rename",
"drop_column",
"add_column",
"drop_constraint",
"add_constraint",
"not_null_action",
"index_action",
"alter_size",
"execute",
"executemany",
"update",
"insert",
"delete",
"select",
}
operation = call.func.attr
is_table_handler = (
isinstance(call.func.value, ast.Attribute)
and call.func.value.attr == "TableHandler"
and isinstance(call.func.value.value, ast.Name)
and call.func.value.value.id == "backend"
)
if operation not in interesting and not is_table_handler:
operation = call.func.attr
is_table_handler = self._is_table_handler_call(call)
if operation not in INTERESTING_OPS and not is_table_handler:
return None
tables = self.find_tables(call)
if (
isinstance(call.func, ast.Attribute)
and call.func.attr == "table_rename"
and isinstance(call.func.value, ast.Attribute)
and isinstance(call.func.value.value, ast.Name)
and call.func.value.value.id == "backend"
and isinstance(call.args, list)
):
for arg in call.args[:2]:
if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
table = arg.value
elif isinstance(arg, ast.Name):
table = self.table_vars.get(arg.id)
elif (
isinstance(arg, ast.Attribute)
and isinstance(arg.value, ast.Name)
and arg.value.id == "cls"
and arg.attr == "_table"
):
table = self.model_names.get(self.current_class)
if table:
table = table.replace(".", "_")
else:
table = None
tables += self._extra_tables_from_table_rename(call, tables)
if table and table not in tables:
tables.append(table)
# Wer wird aufgerufen?
if isinstance(call.func.value, ast.Name):
obj = call.func.value.id
else:
obj = ast.unparse(call.func.value)
args = []
for arg in call.args:
try:
args.append(ast.unparse(arg))
except Exception:
args.append("<unknown>")
kwargs = {}
for kw in call.keywords:
try:
kwargs[kw.arg] = ast.unparse(kw.value)
except Exception:
kwargs[kw.arg] = "<unknown>"
code = ast.get_source_segment(self.source, call)
for name, value in self.alias_vars.items():
code = re.sub(
rf"\b{re.escape(name)}\b",
f"({value})",
code,
)
code = self._render_code(call)
return {
"object": obj,
"operation": operation,
@@ -409,8 +439,69 @@ class RegisterExtractor(ast.NodeVisitor):
"code": code,
}
def analyse_file(module, modulepath, filename):
def _extra_tables_from_table_rename(self, call, existing_tables):
"""
Sonderfall `backend.TableHandler.table_rename(old, new)`: die ersten
beiden Argumente können selbst Tabellen(-namen) sein und werden hier
zusätzlich zu den bereits gefundenen Tabellen ergänzt.
"""
is_table_rename = (
call.func.attr == "table_rename"
and isinstance(call.func.value, ast.Attribute)
and isinstance(call.func.value.value, ast.Name)
and call.func.value.value.id == "backend"
and isinstance(call.args, list)
)
if not is_table_rename:
return []
extra = []
for arg in call.args[:2]:
if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
table = arg.value
elif isinstance(arg, ast.Name):
table = self.table_vars.get(arg.id)
elif (
isinstance(arg, ast.Attribute)
and isinstance(arg.value, ast.Name)
and arg.value.id == "cls"
and arg.attr == "_table"
):
table = self.model_names.get(self.current_class)
if table:
table = table.replace(".", "_")
else:
table = None
if table and table not in existing_tables and table not in extra:
extra.append(table)
return extra
def _render_code(self, call):
"""
Quelltext des Aufrufs, wobei zuvor gemerkte Alias-Variablen
(siehe alias_vars) durch ihren ursprünglichen Ausdruck ersetzt
werden, damit der Code auch ohne Kontext lesbar ist.
"""
code = ast.get_source_segment(self.source, call)
for name, value in self.alias_vars.items():
code = re.sub(
rf"\b{re.escape(name)}\b",
f"({value})",
code,
)
return code
# --------------------------------------------------------------------------
# Datei-/Repo-Ebene
# --------------------------------------------------------------------------
def analyse_file(module, modulepath, filename):
"""Parst eine einzelne Datei und liefert alle __register__-Ergebnisse mit Operationen."""
path = modulepath / filename
try:
source = path.read_text(encoding="utf-8")
@@ -422,11 +513,11 @@ def analyse_file(module, modulepath, filename):
visitor = RegisterExtractor(source, module, filename)
visitor.visit(tree)
results = [r for r in visitor.results if r["operations"]]
return results
return [r for r in visitor.results if r["operations"]]
def detect_version(repo: Path) -> str:
"""Liest die aktuelle Version aus trytond/CHANGELOG."""
changelog = repo / "trytond" / "CHANGELOG"
if not changelog.exists():
@@ -435,49 +526,78 @@ def detect_version(repo: Path) -> str:
with changelog.open("r", encoding="utf-8") as f:
for line in f:
if line.startswith("Version "):
return line.split()[1].rsplit('.', 1)[0]
return line.split()[1].rsplit(".", 1)[0]
return "unknown"
def find_module(repo: Path, module: str) -> Path | None:
def find_module(repo: Path, module: str) -> Path | None:
"""Sucht ein Modul entweder unter modules/ oder trytond/trytond/."""
candidates = [
repo / "modules" / module,
repo / "trytond" / "trytond" / module,
]
for c in candidates:
if c.exists():
return c
for candidate in candidates:
if candidate.exists():
return candidate
return None
def main():
def analyse_module(repo: Path, module: str) -> list | None:
"""Analysiert alle .py-Dateien eines Moduls. None, falls Modul nicht gefunden."""
module_path = find_module(repo, module)
if not module_path:
return None
results = []
for py in sorted(module_path.rglob("*.py")):
filename = str(py.relative_to(module_path))
results.extend(analyse_file(module, module_path, filename))
return results
def write_results(output: Path, version: str, module: str, results: list) -> Path:
"""Schreibt die Ergebnisse eines Moduls als JSON und gibt den Pfad zurück."""
outdir = output / version
outdir.mkdir(parents=True, exist_ok=True)
outfile = outdir / f"{module}.json"
with outfile.open("w", encoding="utf-8") as f:
json.dump(
{
"version": version,
"module": module,
"files": results,
},
f,
indent=2,
)
return outfile
# --------------------------------------------------------------------------
# CLI
# --------------------------------------------------------------------------
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument(
"--repo",
help="Path to Tryton repository",
)
parser.add_argument("--repo", help="Path to Tryton repository")
parser.add_argument("--modules", nargs="+", help="Module names")
parser.add_argument("--output", help="Output directory")
parser.add_argument(
"--modules",
nargs="+",
help="Module names",
)
return parser.parse_args()
parser.add_argument(
"--output",
help="Output directory",
)
args = parser.parse_args()
def main():
args = parse_args()
repo = Path(args.repo or DEFAULT_REPO)
modules = args.modules or DEFAULT_MODULES
output = Path(args.output or DEFAULT_OUTPUT)
version = detect_version(repo)
@@ -488,35 +608,13 @@ def main():
print()
for module in modules:
results = analyse_module(repo, module)
module_path = find_module(repo, module)
if not module_path:
if results is None:
print(f"Module not found: {module}")
continue
results = []
for py in sorted(module_path.rglob("*.py")):
filename = str(py.relative_to(module_path))
results.extend(analyse_file(module, module_path, filename))
outdir = output / version
outdir.mkdir(parents=True, exist_ok=True)
outfile = outdir / f"{module}.json"
with outfile.open("w", encoding="utf-8") as f:
json.dump(
{
"version": version,
"module": module,
"files": results,
},
f,
indent=2,
)
outfile = write_results(output, version, module, results)
print(f"{module:25} -> {outfile}")