-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathschema.py
More file actions
118 lines (98 loc) · 4.15 KB
/
schema.py
File metadata and controls
118 lines (98 loc) · 4.15 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
""" NOTE: this file represents the schema parsing logic for jsonschema_gentypes
"""
import random
import string
import logging
import contextlib
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:
# add three random letters for uniqueness
root += random.choice(string.ascii_letters).upper()
root += random.choice(string.ascii_letters).upper()
root += random.choice(string.ascii_letters).upper()
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.error(f"Error when generating schema type: {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) as f:
output = f.read()
return output
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")