|
from enum import Enum |
|
from typing import List |
|
from typing import Any, Dict |
|
|
|
import jsonschema |
|
from jsf import JSF |
|
from pydantic import BaseModel, Field |
|
|
|
from app.core.errors import VendorError |
|
from app.schemas.requests import Attribute |
|
from app.utils.converter import to_snake_case |
|
from app.utils.logger import exception_to_str |
|
|
|
|
|
def validate_json_data(data: Dict[str, Any], schema: Dict[str, Any]): |
|
""" |
|
Standalone JSON schema validation utility |
|
""" |
|
try: |
|
jsonschema.validate(instance=data, schema=schema) |
|
except jsonschema.ValidationError as e: |
|
raise VendorError(f"Vendor generated invalid data {exception_to_str(e)}") |
|
|
|
|
|
def validate_json_schema(schema: Dict[str, Any]): |
|
""" |
|
Standalone JSON schema validation utility |
|
""" |
|
if schema == {}: |
|
raise ValueError(f"JSON Schema validation failed") |
|
|
|
try: |
|
faker = JSF(schema) |
|
_ = faker.generate() |
|
except: |
|
raise ValueError(f"JSON Schema validation failed") |
|
|
|
|
|
SUPPORTED_DATA_TYPE = [ |
|
"string", |
|
"int", |
|
"float", |
|
"bool", |
|
"list[string]", |
|
"list[int]", |
|
"list[float]", |
|
"list[bool]", |
|
] |
|
|
|
|
|
def convert_attribute_to_model(attributes: Dict[str, Attribute]) -> Dict[str, Any]: |
|
import_code = "" |
|
enum_code_list = [] |
|
master_class_code = "class Product(BaseModel):\n" |
|
for key, value in attributes.items(): |
|
description = value.description |
|
data_type = value.data_type |
|
allowed_values = value.allowed_values |
|
is_list = False |
|
|
|
if data_type not in SUPPORTED_DATA_TYPE: |
|
raise ValueError(f"Data type {data_type} is not supported") |
|
|
|
if "list" in data_type: |
|
is_list = True |
|
|
|
if "int" in data_type: |
|
data_type = "int" |
|
elif "float" in data_type: |
|
data_type = "float" |
|
elif "bool" in data_type: |
|
data_type = "bool" |
|
elif "string" in data_type: |
|
data_type = "str" |
|
|
|
if len(allowed_values) > 0: |
|
enum_code = f"class {key.capitalize()}Enum(str, Enum):\n" |
|
for i, allowed_value in enumerate(allowed_values): |
|
enum_name = f'{to_snake_case(allowed_value).upper()}_{i}' |
|
|
|
if "'" in allowed_value: |
|
enum_code += f' E{enum_name} = "{allowed_value}"\n' |
|
else: |
|
enum_code += f" E{enum_name} = '{allowed_value}'\n" |
|
enum_code_list.append(enum_code) |
|
data_type = f"{key.capitalize()}Enum" |
|
|
|
if is_list: |
|
data_type = f"List[{data_type}]" |
|
|
|
if "'" in description: |
|
master_class_code += ( |
|
f' {key}: {data_type} = Field(..., description="{description}")\n' |
|
) |
|
else: |
|
master_class_code += ( |
|
f" {key}: {data_type} = Field(..., description='{description}')\n" |
|
) |
|
|
|
entire_code = import_code + "\n".join(enum_code_list) + "\n" + master_class_code |
|
exec(entire_code, globals()) |
|
|
|
return Product |
|
|