|
|
@ -1,7 +1,7 @@ |
|
|
import re |
|
|
import re |
|
|
import warnings |
|
|
import warnings |
|
|
from collections.abc import Sequence |
|
|
from collections.abc import Sequence |
|
|
from copy import copy, deepcopy |
|
|
from copy import copy |
|
|
from dataclasses import dataclass, is_dataclass |
|
|
from dataclasses import dataclass, is_dataclass |
|
|
from enum import Enum |
|
|
from enum import Enum |
|
|
from functools import lru_cache |
|
|
from functools import lru_cache |
|
|
@ -305,94 +305,12 @@ def get_definitions( |
|
|
if "description" in item_def: |
|
|
if "description" in item_def: |
|
|
item_description = cast(str, item_def["description"]).split("\f")[0] |
|
|
item_description = cast(str, item_def["description"]).split("\f")[0] |
|
|
item_def["description"] = item_description |
|
|
item_def["description"] = item_description |
|
|
new_mapping, new_definitions = _remap_definitions_and_field_mappings( |
|
|
# definitions: dict[DefsRef, dict[str, Any]] |
|
|
model_name_map=model_name_map, |
|
|
# but mypy complains about general str in other places that are not declared as |
|
|
definitions=definitions, # type: ignore[arg-type] |
|
|
# DefsRef, although DefsRef is just str: |
|
|
field_mapping=field_mapping, |
|
|
# DefsRef = NewType('DefsRef', str) |
|
|
) |
|
|
# So, a cast to simplify the types here |
|
|
return new_mapping, new_definitions |
|
|
return field_mapping, cast(dict[str, dict[str, Any]], definitions) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _replace_refs( |
|
|
|
|
|
*, |
|
|
|
|
|
schema: dict[str, Any], |
|
|
|
|
|
old_name_to_new_name_map: dict[str, str], |
|
|
|
|
|
) -> dict[str, Any]: |
|
|
|
|
|
new_schema = deepcopy(schema) |
|
|
|
|
|
for key, value in new_schema.items(): |
|
|
|
|
|
if key == "$ref": |
|
|
|
|
|
value = schema["$ref"] |
|
|
|
|
|
if isinstance(value, str): |
|
|
|
|
|
ref_name = schema["$ref"].split("/")[-1] |
|
|
|
|
|
if ref_name in old_name_to_new_name_map: |
|
|
|
|
|
new_name = old_name_to_new_name_map[ref_name] |
|
|
|
|
|
new_schema["$ref"] = REF_TEMPLATE.format(model=new_name) |
|
|
|
|
|
continue |
|
|
|
|
|
if isinstance(value, dict): |
|
|
|
|
|
new_schema[key] = _replace_refs( |
|
|
|
|
|
schema=value, |
|
|
|
|
|
old_name_to_new_name_map=old_name_to_new_name_map, |
|
|
|
|
|
) |
|
|
|
|
|
elif isinstance(value, list): |
|
|
|
|
|
new_value = [] |
|
|
|
|
|
for item in value: |
|
|
|
|
|
if isinstance(item, dict): |
|
|
|
|
|
new_item = _replace_refs( |
|
|
|
|
|
schema=item, |
|
|
|
|
|
old_name_to_new_name_map=old_name_to_new_name_map, |
|
|
|
|
|
) |
|
|
|
|
|
new_value.append(new_item) |
|
|
|
|
|
|
|
|
|
|
|
else: |
|
|
|
|
|
new_value.append(item) |
|
|
|
|
|
new_schema[key] = new_value |
|
|
|
|
|
return new_schema |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _remap_definitions_and_field_mappings( |
|
|
|
|
|
*, |
|
|
|
|
|
model_name_map: ModelNameMap, |
|
|
|
|
|
definitions: dict[str, Any], |
|
|
|
|
|
field_mapping: dict[ |
|
|
|
|
|
tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue |
|
|
|
|
|
], |
|
|
|
|
|
) -> tuple[ |
|
|
|
|
|
dict[tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue], |
|
|
|
|
|
dict[str, Any], |
|
|
|
|
|
]: |
|
|
|
|
|
old_name_to_new_name_map = {} |
|
|
|
|
|
for field_key, schema in field_mapping.items(): |
|
|
|
|
|
model = field_key[0].type_ |
|
|
|
|
|
if model not in model_name_map or "$ref" not in schema: |
|
|
|
|
|
continue |
|
|
|
|
|
new_name = model_name_map[model] |
|
|
|
|
|
old_name = schema["$ref"].split("/")[-1] |
|
|
|
|
|
if old_name in {f"{new_name}-Input", f"{new_name}-Output"}: |
|
|
|
|
|
continue |
|
|
|
|
|
old_name_to_new_name_map[old_name] = new_name |
|
|
|
|
|
|
|
|
|
|
|
new_field_mapping: dict[ |
|
|
|
|
|
tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue |
|
|
|
|
|
] = {} |
|
|
|
|
|
for field_key, schema in field_mapping.items(): |
|
|
|
|
|
new_schema = _replace_refs( |
|
|
|
|
|
schema=schema, |
|
|
|
|
|
old_name_to_new_name_map=old_name_to_new_name_map, |
|
|
|
|
|
) |
|
|
|
|
|
new_field_mapping[field_key] = new_schema |
|
|
|
|
|
|
|
|
|
|
|
new_definitions = {} |
|
|
|
|
|
for key, value in definitions.items(): |
|
|
|
|
|
if key in old_name_to_new_name_map: |
|
|
|
|
|
new_key = old_name_to_new_name_map[key] |
|
|
|
|
|
else: |
|
|
|
|
|
new_key = key |
|
|
|
|
|
new_value = _replace_refs( |
|
|
|
|
|
schema=value, |
|
|
|
|
|
old_name_to_new_name_map=old_name_to_new_name_map, |
|
|
|
|
|
) |
|
|
|
|
|
new_definitions[new_key] = new_value |
|
|
|
|
|
return new_field_mapping, new_definitions |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def is_scalar_field(field: ModelField) -> bool: |
|
|
def is_scalar_field(field: ModelField) -> bool: |
|
|
|