Skip to content

Commit c88dc6a

Browse files
FanouZeng-TTFanouZeng-TT
committed
fix: enforce conditional total amount bounds
1 parent a0d8308 commit c88dc6a

4 files changed

Lines changed: 453 additions & 4 deletions

File tree

‎postprocess_models.py‎

Lines changed: 228 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414

1515
"""Post-generation fixes for constraints datamodel-code-generator ignores.
1616
17-
Six constraint families are handled:
17+
Seven constraint families are handled:
1818
1919
* ``minProperties`` on an object schema WITH declared properties is dropped by
2020
the generator (issue #49): every field is optional, so an empty instance
@@ -74,6 +74,12 @@
7474
then injects a ``model_validator(mode="after")``. More complex conditions are
7575
skipped rather than approximated.
7676
77+
* Simple conditional numeric bounds are also dropped: well-known ``Total``
78+
categories constrain ``amount`` to negative or non-negative values. The script
79+
accepts only one discriminator, one target field, and one numeric bound in each
80+
``if``/``then`` rule, and injects a ``model_validator(mode="after")``. Rules
81+
with ``else`` or multiple consequence fields are skipped.
82+
7783
* ``additionalProperties: false`` on an object schema with named properties is
7884
normally overridden by the generator's ``--extra-fields=allow`` flag. The
7985
script detects schemas with ``additionalProperties: false`` and flips their
@@ -148,6 +154,33 @@ def {marker}(self):
148154
return self
149155
'''
150156

157+
_CONDITIONAL_NUMERIC_MARKER = "_enforce_conditional_numeric_bounds"
158+
159+
_CONDITIONAL_NUMERIC_TEMPLATE = '''
160+
@model_validator(mode="after")
161+
def {marker}(self):
162+
"""JSON Schema if/then: enforce conditional numeric bounds."""
163+
rules = {rules!r}
164+
operators = {{
165+
"minimum": lambda value, bound: value >= bound,
166+
"exclusiveMinimum": lambda value, bound: value > bound,
167+
"maximum": lambda value, bound: value <= bound,
168+
"exclusiveMaximum": lambda value, bound: value < bound,
169+
}}
170+
for rule in rules:
171+
if getattr(self, rule["discriminator"], None) not in rule["values"]:
172+
continue
173+
value = getattr(self, rule["field"], None)
174+
if value is not None and not operators[rule["bound"]](
175+
value, rule["value"]
176+
):
177+
raise ValueError(
178+
f"Field {{rule['field']!r}} violates conditional "
179+
f"{{rule['bound']}}={{rule['value']}}"
180+
)
181+
return self
182+
'''
183+
151184
_UNIQUE_VALIDATOR_TEMPLATE = '''
152185
@field_validator("{field}", mode="after")
153186
def {marker}_{field}(cls, value): # noqa: N805
@@ -690,6 +723,162 @@ def inject_conditional_required(source, class_name, rules):
690723
return _ensure_pydantic_import(out, "model_validator")
691724

692725

726+
def find_conditional_numeric_bounds(schema_dir):
727+
"""Map generated classes to simple if/then numeric-bound rules."""
728+
rules_by_class = {}
729+
bound_names = {
730+
"minimum",
731+
"exclusiveMinimum",
732+
"maximum",
733+
"exclusiveMaximum",
734+
}
735+
736+
def describe(node, properties):
737+
if not isinstance(node, dict) or set(node) != {"if", "then"}:
738+
return None
739+
condition = node["if"]
740+
consequence = node["then"]
741+
if (
742+
not isinstance(condition, dict)
743+
or set(condition) != {"properties", "required"}
744+
or not isinstance(consequence, dict)
745+
or set(consequence) != {"properties"}
746+
):
747+
return None
748+
condition_props = condition["properties"]
749+
condition_required = condition["required"]
750+
consequence_props = consequence["properties"]
751+
if (
752+
not isinstance(condition_props, dict)
753+
or len(condition_props) != 1
754+
or not isinstance(condition_required, list)
755+
or len(condition_required) != 1
756+
or not isinstance(consequence_props, dict)
757+
or len(consequence_props) != 1
758+
):
759+
return None
760+
discriminator, predicate = next(iter(condition_props.items()))
761+
field, constraint = next(iter(consequence_props.items()))
762+
if condition_required != [discriminator] or not isinstance(
763+
predicate, dict
764+
):
765+
return None
766+
if set(predicate) == {"const"}:
767+
values = [predicate["const"]]
768+
elif (
769+
set(predicate) == {"enum"}
770+
and isinstance(predicate["enum"], list)
771+
and predicate["enum"]
772+
):
773+
values = predicate["enum"]
774+
else:
775+
return None
776+
if (
777+
not isinstance(constraint, dict)
778+
or len(constraint) != 1
779+
or not set(constraint) <= bound_names
780+
):
781+
return None
782+
bound, value = next(iter(constraint.items()))
783+
if (
784+
discriminator not in properties
785+
or field not in properties
786+
or isinstance(value, bool)
787+
or not isinstance(value, (int, float))
788+
or any(
789+
not isinstance(item, (str, int, float, bool)) for item in values
790+
)
791+
):
792+
return None
793+
return {
794+
"discriminator": discriminator,
795+
"values": values,
796+
"field": field,
797+
"bound": bound,
798+
"value": value,
799+
}
800+
801+
def walk(node, current_class_name, class_properties, path_str):
802+
if not isinstance(node, dict):
803+
return
804+
if isinstance(node.get("title"), str):
805+
current_class_name = _alias_name(node["title"])
806+
properties = node.get("properties")
807+
if isinstance(properties, dict):
808+
class_properties = properties
809+
if "if" in node or "then" in node:
810+
consequence = node.get("then")
811+
consequence_props = (
812+
consequence.get("properties")
813+
if isinstance(consequence, dict)
814+
else None
815+
)
816+
is_numeric_rule = isinstance(consequence_props, dict) and any(
817+
isinstance(constraint, dict)
818+
and bool(set(constraint) & bound_names)
819+
for constraint in consequence_props.values()
820+
)
821+
if is_numeric_rule:
822+
rule = (
823+
None if "else" in node else describe(node, class_properties)
824+
)
825+
if rule is None:
826+
sys.stderr.write(
827+
f" ! {path_str}: unsupported conditional numeric "
828+
"rule; skipped\n"
829+
)
830+
elif current_class_name is not None:
831+
rules_by_class.setdefault(current_class_name, []).append(
832+
rule
833+
)
834+
if isinstance(properties, dict):
835+
for name, prop in properties.items():
836+
walk(prop, _to_camel_case(name), properties, path_str)
837+
defs = node.get("$defs")
838+
if isinstance(defs, dict):
839+
for def_name, def_node in defs.items():
840+
walk(def_node, _to_camel_case(def_name), {}, path_str)
841+
for key in ("allOf", "anyOf", "oneOf"):
842+
if isinstance(node.get(key), list):
843+
for item in node[key]:
844+
walk(item, current_class_name, class_properties, path_str)
845+
846+
for path in sorted(Path(schema_dir).rglob("*.json")):
847+
try:
848+
schema = json.loads(path.read_text(encoding="utf-8"))
849+
except (OSError, json.JSONDecodeError):
850+
continue
851+
if not isinstance(schema, dict):
852+
continue
853+
root_title = schema.get("title")
854+
initial_class = (
855+
_alias_name(root_title) if root_title else _to_camel_case(path.stem)
856+
)
857+
walk(schema, initial_class, {}, str(path))
858+
return rules_by_class
859+
860+
861+
def inject_conditional_numeric_bounds(source, class_name, rules):
862+
"""Inject simple conditional numeric checks into one generated class."""
863+
class_re = re.compile(rf"^class {re.escape(class_name)}\(", re.M)
864+
match = class_re.search(source)
865+
if not match:
866+
return source
867+
tail = re.compile(r"^\S", re.M)
868+
end_match = tail.search(source, match.end())
869+
end = end_match.start() if end_match else len(source)
870+
if f"def {_CONDITIONAL_NUMERIC_MARKER}(" in source[match.start() : end]:
871+
return source
872+
method = _CONDITIONAL_NUMERIC_TEMPLATE.format(
873+
marker=_CONDITIONAL_NUMERIC_MARKER,
874+
rules=rules,
875+
)
876+
body = source[:end].rstrip("\n")
877+
rest = source[end:]
878+
out = body + "\n" + method + ("\n" + rest if rest else "")
879+
return _ensure_pydantic_import(out, "model_validator")
880+
881+
693882
def find_unique_items_fields(schema_dir):
694883
"""Map generated class names to fields carrying ``uniqueItems``.
695884
@@ -982,6 +1171,41 @@ def _patch_conditional_required():
9821171
return patched, 0
9831172

9841173

1174+
def _patch_conditional_numeric_bounds():
1175+
"""Inject conditional numeric validators; return counts and status."""
1176+
rules_by_class = find_conditional_numeric_bounds(RAW_SCHEMA_DIR)
1177+
if not rules_by_class:
1178+
sys.stdout.write(
1179+
"postprocess: no simple conditional numeric rules found\n"
1180+
)
1181+
return 0, 0
1182+
patched = 0
1183+
for class_name, rules in sorted(rules_by_class.items()):
1184+
hits = []
1185+
for path in sorted(OUTPUT_DIR.rglob("*.py")):
1186+
source = path.read_text(encoding="utf-8")
1187+
if not re.search(
1188+
rf"^class {re.escape(class_name)}\(", source, re.M
1189+
):
1190+
continue
1191+
updated = inject_conditional_numeric_bounds(
1192+
source, class_name, rules
1193+
)
1194+
if updated != source:
1195+
path.write_text(updated, encoding="utf-8")
1196+
patched += 1
1197+
hits.append(path)
1198+
label = (
1199+
", ".join(str(path) for path in hits) or "NO GENERATED CLASS FOUND"
1200+
)
1201+
sys.stdout.write(
1202+
f" conditional numeric bounds on '{class_name}' -> {label}\n"
1203+
)
1204+
if not hits:
1205+
return patched, 1
1206+
return patched, 0
1207+
1208+
9851209
def _patch_unique_items():
9861210
"""Inject uniqueItems validators; return (patched_count, exit_code)."""
9871211
unique_fields_by_class = find_unique_items_fields(SCHEMA_DIR)
@@ -1121,18 +1345,20 @@ def main():
11211345
patched_pn, rc_pn = _patch_property_names()
11221346
patched_ac, rc_ac = _patch_array_contains()
11231347
patched_cr, rc_cr = _patch_conditional_required()
1348+
patched_cn, rc_cn = _patch_conditional_numeric_bounds()
11241349
patched_ui, rc_ui = _patch_unique_items()
11251350
patched_ef, rc_ef = _patch_extra_forbid()
11261351
total = (
11271352
patched_mp
11281353
+ patched_pn
11291354
+ patched_ac
11301355
+ patched_cr
1356+
+ patched_cn
11311357
+ patched_ui
11321358
+ patched_ef
11331359
)
11341360
sys.stdout.write(f"postprocess: {total} module(s) patched\n")
1135-
return rc_mp or rc_pn or rc_ac or rc_cr or rc_ui or rc_ef
1361+
return rc_mp or rc_pn or rc_ac or rc_cr or rc_cn or rc_ui or rc_ef
11361362

11371363

11381364
if __name__ == "__main__":

‎src/ucp_sdk/models/schemas/shopping/types/total.py‎

Lines changed: 39 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818

1919
from __future__ import annotations
2020

21-
from pydantic import BaseModel, ConfigDict
21+
from pydantic import BaseModel, ConfigDict, model_validator
2222

2323
from . import signed_amount
2424

@@ -40,3 +40,41 @@ class Total(BaseModel):
4040
Text to display against the amount. Should reflect appropriate method (e.g., 'Shipping', 'Delivery').
4141
"""
4242
amount: signed_amount.SignedAmount
43+
44+
@model_validator(mode="after")
45+
def _enforce_conditional_numeric_bounds(self):
46+
"""JSON Schema if/then: enforce conditional numeric bounds."""
47+
rules = [
48+
{
49+
"discriminator": "type",
50+
"values": ["discount", "items_discount"],
51+
"field": "amount",
52+
"bound": "exclusiveMaximum",
53+
"value": 0,
54+
},
55+
{
56+
"discriminator": "type",
57+
"values": ["subtotal", "fulfillment", "tax", "fee"],
58+
"field": "amount",
59+
"bound": "minimum",
60+
"value": 0,
61+
},
62+
]
63+
operators = {
64+
"minimum": lambda value, bound: value >= bound,
65+
"exclusiveMinimum": lambda value, bound: value > bound,
66+
"maximum": lambda value, bound: value <= bound,
67+
"exclusiveMaximum": lambda value, bound: value < bound,
68+
}
69+
for rule in rules:
70+
if getattr(self, rule["discriminator"], None) not in rule["values"]:
71+
continue
72+
value = getattr(self, rule["field"], None)
73+
if value is not None and not operators[rule["bound"]](
74+
value, rule["value"]
75+
):
76+
raise ValueError(
77+
f"Field {rule['field']!r} violates conditional "
78+
f"{rule['bound']}={rule['value']}"
79+
)
80+
return self

‎src/ucp_sdk/models/schemas/shopping/types/totals.py‎

Lines changed: 45 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,13 @@
2020

2121
from typing import Annotated
2222

23-
from pydantic import BaseModel, ConfigDict, Field, AfterValidator
23+
from pydantic import (
24+
BaseModel,
25+
ConfigDict,
26+
Field,
27+
AfterValidator,
28+
model_validator,
29+
)
2430
from typing_extensions import TypeAliasType
2531

2632
from . import signed_amount
@@ -51,6 +57,44 @@ class Total(Total_1):
5157
Optional itemized breakdown. The parent entry is always rendered; lines are supplementary. Sum of line amounts MUST equal the parent entry amount.
5258
"""
5359

60+
@model_validator(mode="after")
61+
def _enforce_conditional_numeric_bounds(self):
62+
"""JSON Schema if/then: enforce conditional numeric bounds."""
63+
rules = [
64+
{
65+
"discriminator": "type",
66+
"values": ["discount", "items_discount"],
67+
"field": "amount",
68+
"bound": "exclusiveMaximum",
69+
"value": 0,
70+
},
71+
{
72+
"discriminator": "type",
73+
"values": ["subtotal", "fulfillment", "tax", "fee"],
74+
"field": "amount",
75+
"bound": "minimum",
76+
"value": 0,
77+
},
78+
]
79+
operators = {
80+
"minimum": lambda value, bound: value >= bound,
81+
"exclusiveMinimum": lambda value, bound: value > bound,
82+
"maximum": lambda value, bound: value <= bound,
83+
"exclusiveMaximum": lambda value, bound: value < bound,
84+
}
85+
for rule in rules:
86+
if getattr(self, rule["discriminator"], None) not in rule["values"]:
87+
continue
88+
value = getattr(self, rule["field"], None)
89+
if value is not None and not operators[rule["bound"]](
90+
value, rule["value"]
91+
):
92+
raise ValueError(
93+
f"Field {rule['field']!r} violates conditional "
94+
f"{rule['bound']}={rule['value']}"
95+
)
96+
return self
97+
5498

5599
def _enforce_contains_totals(value):
56100
"""JSON Schema contains/minContains/maxContains (see #49)."""

0 commit comments

Comments
 (0)