sunank200 commented on issue #43378:
URL: https://github.com/apache/airflow/issues/43378#issuecomment-2809042132
Thanks for feedback @pierrejeambrun
I have updated the script now to following:
```
import sys
import json
import yaml
import requests
import re
from deepdiff import DeepDiff
def load_spec_from_file(filepath):
with open(filepath, "r") as f:
if filepath.endswith(".json"):
return json.load(f)
else:
return yaml.safe_load(f)
def load_spec_from_url(url):
response = requests.get(url)
response.raise_for_status()
if url.endswith(".json"):
return response.json()
else:
return yaml.safe_load(response.text)
def normalize_spec(spec, version_prefixes=("/api/v1", "/api/v2")):
new_paths = {}
for path, methods in spec.get("paths", {}).items():
for prefix in version_prefixes:
if path.startswith(prefix):
path = path[len(prefix):]
new_paths[path or "/"] = methods
spec["paths"] = new_paths
spec.pop("servers", None)
return spec
def summarize_diff(diff):
lines = []
if "iterable_item_added" in diff:
for key, value in diff["iterable_item_added"].items():
if "parameters" in key and isinstance(value, dict):
param_name = value.get("name", "unknown")
lines.append(f"• Parameter '{param_name}' added")
else:
simple_key = re.sub(r"root\[(.*?)\]", r"\1", key)
lines.append(f"• Added {simple_key}: {value}")
if "dictionary_item_removed" in diff:
for key in diff["dictionary_item_removed"]:
if "parameters" in key:
lines.append("• A parameter was removed")
else:
simple_key = re.sub(r"root\[(.*?)\]", r"\1", key)
lines.append(f"• Removed {simple_key}")
if "values_changed" in diff:
for key, change in diff["values_changed"].items():
if "parameters" in key:
p_old = change.get("old_value", {})
p_new = change.get("new_value", {})
if not (isinstance(p_old, dict) and isinstance(p_new, dict)):
lines.append(f"• Changed parameter diff: from {p_old} to
{p_new}")
continue
param_name = p_old.get("name") or p_new.get("name") or
"unknown"
old_schema = p_old.get("schema", {})
new_schema = p_new.get("schema", {})
old_type = old_schema.get("type") if isinstance(old_schema,
dict) else None
new_type = new_schema.get("type") if isinstance(new_schema,
dict) else None
old_default = old_schema.get("default") if
isinstance(old_schema, dict) else None
new_default = new_schema.get("default") if
isinstance(new_schema, dict) else None
details = []
if old_type != new_type:
details.append(f"schema type from '{old_type}' to
'{new_type}'")
if old_default != new_default:
details.append(f"default from '{old_default}' to
'{new_default}'")
detail_str = ", ".join(details)
if detail_str:
lines.append(f"• Parameter '{param_name}' changed:
{detail_str}")
else:
lines.append(f"• Parameter '{param_name}' changed")
else:
simple_key = re.sub(r"root\[(.*?)\]", r"\1", key)
old_value = change.get("old_value")
new_value = change.get("new_value")
def simple_repr(val):
if isinstance(val, (dict, list)):
return f"<{type(val).__name__}>"
return val
lines.append(f"• Changed {simple_key}: from
{simple_repr(old_value)} to {simple_repr(new_value)}")
return "\n".join(lines)
def compare_specs(spec1, spec2):
breaking_changes = []
additional_notes = []
paths1 = spec1.get("paths", {})
paths2 = spec2.get("paths", {})
exclude_patterns = [
r".*\['description'\].*",
r".*\['summary'\].*",
r".*\['security'\].*",
r".*\['x\-openapi\-router\-controller'\].*",
r"root\['responses'\]\['4\d\d'\].*",
r".*\['example'\].*"
]
for path in sorted(paths1.keys()):
if path not in paths2:
breaking_changes.append(f"Endpoint '{path}' removed in new spec")
else:
for method in sorted(paths1[path].keys()):
if method not in paths2[path]:
breaking_changes.append(f"Method '{method.upper()}' for
endpoint '{path}' removed in new spec")
else:
diff = DeepDiff(
paths1[path][method],
paths2[path][method],
ignore_order=True,
exclude_regex_paths=exclude_patterns
)
if diff:
summary = summarize_diff(diff)
if summary.strip():
breaking_changes.append(f"Method
'{method.upper()}' for endpoint '{path}' changed:\n{summary}")
for path in sorted(paths2.keys()):
if path not in paths1:
additional_notes.append(f"New endpoint '{path}' added in new
spec")
else:
for method in sorted(paths2[path].keys()):
if method not in paths1[path]:
additional_notes.append(f"New method '{method.upper()}'
for endpoint '{path}' added in new spec")
return breaking_changes, sorted(additional_notes)
if __name__ == "__main__":
if len(sys.argv) < 3:
print("Usage: python compare_openapi.py <spec1_path_or_url>
<spec2_path_or_url>")
sys.exit(1)
spec1_source = sys.argv[1]
spec2_source = sys.argv[2]
try:
if spec1_source.startswith("http"):
spec1 = load_spec_from_url(spec1_source)
else:
spec1 = load_spec_from_file(spec1_source)
except Exception as e:
print(f"Failed to load spec from {spec1_source}: {e}")
sys.exit(1)
try:
if spec2_source.startswith("http"):
spec2 = load_spec_from_url(spec2_source)
else:
spec2 = load_spec_from_file(spec2_source)
except Exception as e:
print(f"Failed to load spec from {spec2_source}: {e}")
sys.exit(1)
spec1 = normalize_spec(spec1)
spec2 = normalize_spec(spec2)
breaking, notes = compare_specs(spec1, spec2)
report_lines = []
report_lines.append("API Comparison Report:")
report_lines.append("======================\n")
if notes:
report_lines.append("Additional Notes (Non-Breaking Changes):")
for note in notes:
report_lines.append(f" - {note}")
report_lines.append("")
if breaking:
report_lines.append("Breaking Changes Detected:")
for change in breaking:
report_lines.append(f"\n - {change}")
else:
report_lines.append("No breaking changes detected.")
report_content = "\n".join(report_lines)
print(report_content)
output_filename = "report.txt"
try:
with open(output_filename, "w") as f:
f.write(report_content)
print(f"\nReport written to {output_filename}")
except Exception as e:
print(f"Failed to write report to {output_filename}: {e}")
```
Following is the report generated now which is more readable:
[report.txt](https://github.com/user-attachments/files/19774860/report.txt)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
