Coverage for formkit_ninja / parser / schema_walker.py: 20.00%

42 statements  

« prev     ^ index     » next       coverage.py v7.13.1, created at 2026-02-20 04:40 +0000

1""" 

2Schema traversal utilities for FormKit schemas. 

3 

4This module provides SchemaWalker, a focused traversal helper used to collect 

5NodePath instances in a consistent order. 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import List, Protocol, Union 

11 

12from formkit_ninja.formkit_schema import FormKitSchema 

13from formkit_ninja.parser.generator_config import GeneratorConfig 

14from formkit_ninja.parser.type_convert import NodePath 

15 

16SchemaInput = Union[List[dict], FormKitSchema] 

17 

18 

19class SchemaVisitor(Protocol): 

20 """Protocol for SchemaWalker visitors.""" 

21 

22 def on_node(self, node_path: NodePath) -> None: 

23 """Handle a NodePath during traversal.""" 

24 

25 

26class SchemaWalker: 

27 """Traverse FormKit schemas and collect NodePath instances.""" 

28 

29 def __init__(self, config: GeneratorConfig) -> None: 

30 self.config = config 

31 

32 def collect_nodepaths( 

33 self, 

34 schema: SchemaInput, 

35 *, 

36 abstract_base_info: dict[str, bool] | None = None, 

37 ) -> List[NodePath]: 

38 """Collect NodePath instances from a schema in pre-order.""" 

39 nodepaths: List[NodePath] = [] 

40 shared_abstract_info = abstract_base_info or {} 

41 

42 schema_dicts: List[dict] = [] 

43 if isinstance(schema, FormKitSchema): 

44 for node in schema.__root__: 

45 if isinstance(node, str): 

46 continue 

47 schema_dicts.append(node.dict(exclude_none=True)) 

48 else: 

49 schema_dicts = schema 

50 

51 def traverse_node(node_dict: dict, parent_path: NodePath | None = None) -> None: 

52 temp_path = self.config.node_path_class.from_obj(node_dict) 

53 node = temp_path.node 

54 

55 if parent_path is not None: 

56 node_path = parent_path / node 

57 else: 

58 node_path = self.config.node_path_class(node) 

59 

60 node_path._config = self.config 

61 node_path._abstract_base_info = shared_abstract_info 

62 nodepaths.append(node_path) 

63 

64 children = node_dict.get("children", []) 

65 if children: 

66 for child_dict in children: 

67 if isinstance(child_dict, str): 

68 continue 

69 traverse_node(child_dict, node_path) 

70 

71 for node_dict in schema_dicts: 

72 traverse_node(node_dict) 

73 

74 return nodepaths 

75 

76 def walk(self, schema: SchemaInput, visitor: SchemaVisitor) -> None: 

77 """Walk a schema and invoke visitor for each NodePath.""" 

78 for node_path in self.collect_nodepaths(schema): 

79 visitor.on_node(node_path)