-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathschema.py
More file actions
166 lines (133 loc) · 5.78 KB
/
schema.py
File metadata and controls
166 lines (133 loc) · 5.78 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
""" NOTE: this file represents the schema parsing logic for jsonschema_gentypes
"""
import logging
import contextlib
import re
from typing import Dict
from jsonschema_gentypes.cli import process_config
from jsonschema_gentypes import configuration
import referencing
import tempfile
import json
import referencing.exceptions
from polyapi.constants import JSONSCHEMA_TO_PYTHON_TYPE_MAP
def _cleanup_input_for_gentypes(input_data: Dict):
""" cleanup input_data in place to make it more suitable for jsonschema_gentypes
"""
for k, v in input_data.items():
if isinstance(v, dict):
_cleanup_input_for_gentypes(v)
elif k == "enum":
# jsonschema_gentypes doesn't like double quotes in enums
# TODO fix this upstream
for idx, enum in enumerate(v):
if isinstance(enum, str):
v[idx] = enum.replace('"', "'")
def _temp_store_input_data(input_data: Dict) -> str:
"""take in the input data and store it in a temporary json file"""
with tempfile.NamedTemporaryFile(
mode="w", delete=False, prefix="polyapi_", suffix=".json"
) as temp_file:
json.dump(input_data, temp_file)
return temp_file.name
def wrapped_generate_schema_types(type_spec: dict, root, fallback_type):
from polyapi.utils import pascalCase
if not root:
root = "List" if fallback_type == "List" else "Dict"
if type_spec.get("x-poly-ref") and type_spec["x-poly-ref"].get("path"):
# x-poly-ref occurs when we have an unresolved reference
# lets name the root after the reference for some level of visibility
root += pascalCase(type_spec["x-poly-ref"]["path"].replace(".", " "))
else:
# if we have no root, just add "My"
root = "My" + root
root = clean_title(root)
try:
return root, generate_schema_types(type_spec, root=root)
except RecursionError:
# some schemas are so huge, our library cant handle it
# TODO identify critical recursion penalty and maybe switch underlying logic to iterative?
return fallback_type, ""
except referencing.exceptions.CannotDetermineSpecification:
# just go with fallback_type here
# we couldn't match the right $ref earlier in resolve_poly_refs
# {'$ref': '#/definitions/FinanceAccountListModel'}
return fallback_type, ""
except:
logging.warning(f"WARNING parsing jsonschema failed: {type_spec}\nusing fallback type '{fallback_type}'")
return fallback_type, ""
def generate_schema_types(input_data: Dict, root=None):
"""takes in a Dict representing a schema as input then appends the resulting python code to the output file"""
_cleanup_input_for_gentypes(input_data)
tmp_input = _temp_store_input_data(input_data)
tmp_output = tempfile.NamedTemporaryFile(
mode="w", delete=False, prefix="polyapi_", suffix=".py"
).name
config: configuration.Configuration = {
"python_version": None, # type: ignore
"generate": [
{
"source": tmp_input,
"destination": tmp_output,
"root_name": root,
"api_arguments": {"get_name_properties": "UpperFirst"},
}
],
}
# jsonschema_gentypes prints source to stdout
# no option to surpress so we do this
with contextlib.redirect_stdout(None):
process_config(config, [tmp_input])
with open(tmp_output, encoding='utf-8') as f:
output = f.read()
output = clean_malformed_examples(output)
return output
# Matches commented example headers emitted by jsonschema-gentypes before a raw
# multiline JSON object/array body that is not commented out.
MALFORMED_EXAMPLE_HEADER_PATTERN = re.compile(
r"^\s*#\s*(?:\|\s*)?example:\s*([\[{])\s*$"
)
# Regex to fix invalid escape sequences in docstrings
INVALID_ESCAPE_PATTERNS = [
# Fix "\ " (backslash space) which is not a valid escape sequence
(re.compile(r'\\(\s)', re.DOTALL), r'\1'),
# Fix other common invalid escape sequences in docstrings
(re.compile(r'\\([^nrtbfav"\'\\])', re.DOTALL), r'\\\\\1'),
]
def clean_malformed_examples(example: str) -> str:
""" there is a bug in the `jsonschmea_gentypes` library where if an example from a jsonchema is an object,
it will break the code because the object won't be properly commented out. Also fixes invalid escape sequences.
"""
cleaned_lines = []
balance = 0
skipping_example = False
for line in example.splitlines(keepends=True):
if not skipping_example:
if MALFORMED_EXAMPLE_HEADER_PATTERN.match(line):
skipping_example = True
balance = line.count("{") + line.count("[") - line.count("}") - line.count("]")
continue
cleaned_lines.append(line)
continue
balance += line.count("{") + line.count("[") - line.count("}") - line.count("]")
if balance <= 0:
skipping_example = False
cleaned_example = "".join(cleaned_lines)
# Fix invalid escape sequences in docstrings
for pattern, replacement in INVALID_ESCAPE_PATTERNS:
cleaned_example = pattern.sub(replacement, cleaned_example)
return cleaned_example
def clean_title(title: str) -> str:
""" used by library generation, sometimes functions can be added with spaces in the title
or other nonsense. fix them!
"""
title = title.replace(" ", "")
# certain reserved words cant be titles, let's replace them
if title == "List":
title = "List_"
return title
def map_primitive_types(type_: str) -> str:
# Define your mapping logic here
return JSONSCHEMA_TO_PYTHON_TYPE_MAP.get(type_, "Any")
def is_primitive(type_: str) -> bool:
return type_ in JSONSCHEMA_TO_PYTHON_TYPE_MAP