Support for processing pseudo-arrays

[Miskler]

Hi! In my projects I needed to implement a handler for “pseudo-arrays”.
Essentially, these are dictionaries where the keys are numbers.
This format appears in a couple of public endpoints, and I ran into it while reverse-engineering the API of one store.

I can prepare a full MR if you're interested in having this functionality enabled by default.
Attaching an example of my implementation :)

class PseudoArrayConverter(SchemaBuilder):
    """
    Specialized SchemaBuilder for converting pseudo-arrays
    (dictionaries with sequential numeric string keys) into patternProperties.
    """

    def __init__(
        self,
        schema_uri: str = "https://json-schema.org/draft/2020-12/schema",
    ) -> None:
        super().__init__()
        # Store raw objects by paths for pseudo-array analysis
        self._raw_at_path: Dict[str, Any] = {}
        # Store all added objects in full
        self._all_objects: List[Any] = []

    # Raw data collection
    # ------------------------------------------------------------------ #
    def add_object(self, obj: Any) -> None:
        """Adds an object for analysis."""
        self._all_objects.append(obj)
        self._collect_raw(obj, "#")
        super().add_object(obj)

    def _collect_raw(self, obj: Any, path: str) -> None:
        """Recursively collects raw data by paths."""
        self._raw_at_path[path] = obj

        if isinstance(obj, dict):
            for k, v in obj.items():
                self._collect_raw(v, f"{path}/{k}")
        elif isinstance(obj, (list, tuple)):
            for i, v in enumerate(obj):
                self._collect_raw(v, f"{path}/{i}")

    # Schema generation with pseudo-array conversion
    # ------------------------------------------------------------------ #
    def to_schema(self) -> Dict[str, Any]:
        """Generates a schema with converted pseudo-arrays."""
        schema = super().to_schema()
        self._convert_pseudo_arrays(schema, "#")
        return schema

    # Pseudo-array conversion logic
    # ------------------------------------------------------------------ #
    def _convert_pseudo_arrays(self, node: Dict[str, Any], path: str) -> None:
        """
        Recursively converts pseudo-arrays into patternProperties.
        Now uses all added objects to determine types correctly.
        """
        if not (node.get("type") == "object" and "properties" in node):
            self._recurse(node, self._convert_pseudo_arrays, path)
            return

        # Get all values for this path from all objects
        values_at_path = self._get_all_values_at_path(path)

        if not values_at_path:
            self._recurse(node, self._convert_pseudo_arrays, path)
            return

        # Check whether all values are dicts with numeric keys
        all_have_numeric_keys = all(
            isinstance(v, dict) and all(isinstance(k, str) and k.isdigit() for k in v.keys())
            for v in values_at_path
            if v is not None
        )

        if not all_have_numeric_keys:
            self._recurse(node, self._convert_pseudo_arrays, path)
            return

        # Collect all numeric keys from all objects
        all_keys = set()
        for value in values_at_path:
            if isinstance(value, dict):
                all_keys.update(value.keys())

        if not all_keys:
            return

        # Check conditions for converting into a pseudo-array
        if self._should_convert_to_pseudo_array(all_keys):
            # Collect nested objects for schema merging
            nested_objects = []
            for value in values_at_path:
                if isinstance(value, dict):
                    for k in sorted(value.keys(), key=int):
                        nested_objects.append(value[k])

            # Create a schema for the nested objects
            item_schema = self._create_schema_for_objects(nested_objects)

            # Convert into patternProperties
            node.clear()
            node.update(
                {
                    "type": "object",
                    "propertyNames": {"pattern": "^[0-9]+$"},
                    "patternProperties": {"^[0-9]+$": item_schema},
                    "additionalProperties": False,
                }
            )
            return

        self._recurse(node, self._convert_pseudo_arrays, path)

    def _get_all_values_at_path(self, path: str) -> List[Any]:
        """Gets all values for the specified path from all objects."""
        values = []

        # If there is a value in _raw_at_path (from the latest object)
        if path in self._raw_at_path:
            values.append(self._raw_at_path[path])

        # Also look through previous objects
        for obj in self._all_objects[:-1]:
            value = self._extract_value_from_path(obj, path)
            if value is not None:
                values.append(value)

        return values

    def _extract_value_from_path(self, obj: Any, path: str) -> Any:
        """Extracts a value by path from an object."""
        if path == "#":
            return obj

        # Skip the initial "#"
        parts = path[2:].split("/") if path.startswith("#/") else path.split("/")

        current = obj
        for part in parts:
            if isinstance(current, dict) and part in current:
                current = current[part]
            elif isinstance(current, (list, tuple)) and part.isdigit():
                idx = int(part)
                if 0 <= idx < len(current):
                    current = current[idx]
                else:
                    return None
            else:
                return None

        return current

    def _should_convert_to_pseudo_array(self, keys: set) -> bool:
        """Determines whether the structure should be converted to a pseudo-array."""
        return all(k.isdigit() for k in keys)

    def _create_schema_for_objects(self, objects: List[Any]) -> Dict[str, Any]:
        """Creates a schema for a list of objects by merging their types."""
        if not objects:
            return {"type": "object"}

        # Use genson to merge types
        from .to_schema_converter import JsonToSchemaConverter

        builder = JsonToSchemaConverter()
        for obj in objects:
            builder.add_object(obj)

        return builder.to_schema()

    # Helpers
    # ------------------------------------------------------------------ #
    def _recurse(self, node: Dict[str, Any], func: callable, path: str) -> None:
        """Recursively traverses a schema."""
        if node.get("type") == "object" and "properties" in node:
            for k, sub in node["properties"].items():
                func(sub, f"{path}/{k}")

        if node.get("type") == "array":
            items = node.get("items", {})
            if isinstance(items, dict):
                func(items, f"{path}/0")
            elif isinstance(items, list):
                for i, sub in enumerate(items):
                    func(sub, f"{path}/{i}")

        if "anyOf" in node:
            for sub in node["anyOf"]:
                func(sub, path)

Output:

    "orders": {
        "0": {"order_id": 1001, "amount": 299.99, "status": "completed", "t": "john.doe@example.com"},
        "1": {"order_id": 1002, "amount": 149.50, "status": "pending", "t": [{"da": [1]}]},
        "3": {"order_id": 1004, "amount": 89.00,  "status": "completed", "t": {"da": [1]}},
        "4": {"order_id": 1004, "amount": 89.00,  "status": "completed", "t": {}},
    }

    "orders": {
      "type": "object",
      "propertyNames": {
        "pattern": "^[0-9]+$"
      },
      "patternProperties": {
        "^[0-9]+$": {
          "$schema": "http://json-schema.org/schema#",
          "type": "object",
          "properties": {
            "order_id": {
              "type": "integer"
            },
            "amount": {
              "type": "number"
            },
            "status": {
              "type": "string"
            },
            "t": {
              "anyOf": [
                {
                  "type": "string"
                },
                {
                  "type": "array",
                  "items": {
                    "type": "object",
                    "properties": {
                      "da": {
                        "type": "array",
                        "items": {
                          "type": "integer"
                        }
                      }
                    },
                    "required": [
                      "da"
                    ]
                  }
                },
                {
                  "type": "object",
                  "properties": {
                    "da": {
                      "type": "array",
                      "items": {
                        "type": "integer"
                      }
                    }
                  }
                }
              ]
            }
          },
          "required": [
            "amount",
            "order_id",
            "status",
            "t"
          ]
        }
      },
      "additionalProperties": false
    },

[wolverdude]

Interesting. Thanks for sharing this use-case!

I'm curious why you don't just use a seed schema like in this example?

If you want to be able to detect them wherever they show up, it might be simpler to just extend the Object SchemaStrategy to pre-add the pattern property on __init__. That way GenSON will do all the merging logic for you automatically as it adds new items, which should save on memory, compute, and lines-of-code.

[Miskler]

Interesting. Thanks for sharing this use-case!

I'm curious why you don't just use a seed schema like in this example?

If you want to be able to detect them wherever they show up, it might be simpler to just extend the Object SchemaStrategy to pre-add the pattern property on __init__. That way GenSON will do all the merging logic for you automatically as it adds new items, which should save on memory, compute, and lines-of-code.

Because in this example, validation takes place and not the definition of a pseudo-array, that is, if you do not manually specify that there may be a pseudo-array, the code will always assume that there are ordinary keys.

[Miskler]

The version that I sent is very outdated, I have been trying to achieve clean behavior for a very long time, but I came across the fact that the processing architecture does not assume some of the things that I needed (or I do not understand how to use them correctly)

[Miskler]

I found my last note on this, and I tried to write out a technical assignment for myself separately from the code.

We must correctly identify when data is not a dictionary but only a pseudo-array.

How similar patterns are determined: In `add_object`, we already have data that we can use to run our patterns through. When merging schemas, we take the "examples" attribute from our pattern database and run them for the patterns in the provided schema. Consider that if a schema pattern passes the examples of our pattern, then the schema pattern is equivalent. Then, after obtaining a dataset (from examples or real data), we look for the strictest pattern, while combining data (to find areas with common patterns, even if less strict than individually).

Mandatory fields in the final schema:

1. A comment explaining what this pattern is.
2. A list of discarded patterns (discarded patterns are only those that failed selection based on instance data; if they simply have lower priority, they are not included). In subsequent iterations of the algorithm, when merging via `add_schema`, this blacklist must be considered when selecting a pattern or its absence.

Rules:

1. Patterns are created only for JSON data when converting to a schema (except when merging with an existing schema that indicates there is no pattern). During generation from JSON, all empty dictionaries (if there is no case where they have content) should be marked as empty.

2. When merging, we try to find common patterns where possible. In such cases, it is allowed to reduce the strictness of the common pattern if it exists in the list of available patterns.

3. If certain patterns cannot be merged, we place them in `oneOf`.

4. If, when merging patterns, we have an empty candidate (a marked empty dictionary), we simply discard it.

5. If, when merging patterns, we have a non-pattern, non-empty candidate, we discard all patterns.

[Miskler]

The version that I sent is very outdated, I have been trying to achieve clean behavior for a very long time, but I came across the fact that the processing architecture does not assume some of the things that I needed (or I do not understand how to use them correctly)

I meant that when choosing a layer processing tool in genson, as far as I understand, they prohibit parallel processing by multiple handlers natively, i.e. there is a "layer commander" (in most cases, your library has a dictionary class) this makes it difficult to monitor the fulfillment of all requirements (they are actually quite independent of each other)

At the same time, I came up with a concept that could solve this:

# Concept

There is a global manager that aggregates candidates and performs recursive processing.

Processing proceeds as follows: A bottom-up traversal of all schemas/JSON data is performed, and they are provided to comparators in order of priority for processing. Comparators perform the necessary operations and return a layer (they all do not see the schema globally; they work only with their own layer).

Global rules define:
- Priority and a list of comparators

Next, comparators (by priority) are first asked: "Will you process something?"
If the answer is positive, they can return a list of comparators that are NOT ALLOWED to process this layer later.
If the answer was positive, the layer processing function is called. The function always receives:
1. A list of candidate schemas
2. A list of candidate JSON data
3. The result of previous layer processing (if none, then `{}`)
4. The environment path (e.g., `property/anyOf/pattern/property`)

Both lists are already parsed to the required nesting level.

The function returns:
The result of its processing (a dictionary).

[Miskler]

@wolverdude https://github.com/Miskler/genschema?tab=readme-ov-file#-comparison-with-genson

[Miskler]

Finally, I make it!)