[cdd/compound/openapi/utils/emit_utils.py] Fix support for JSON and B…

…igInteger

cdd/compound/gen_utils.py

@@ -392,7 +392,7 @@ def get_functions_and_classes(
         print("\nGenerating: {name!r}".format(name=name))
         or global__all__.append(name_tpl.format(name=name))
         or emitter(
-            get_parser(obj, parse_name)(obj),
+            print("obj:", obj, ";") or get_parser(obj, parse_name)(obj),
             emit_default_doc=emit_default_doc,
             word_wrap=no_word_wrap is None,
             **get_emit_kwarg(decorator_list, emit_call, emit_name, name_tpl, name),

cdd/compound/openapi/utils/emit_utils.py

@@ -17,8 +17,6 @@
     Name,
     Return,
     Store,
-    Subscript,
-    Tuple,
     alias,
     arguments,
     keyword,
@@ -31,17 +29,17 @@
 from os import path
 from platform import system
 
+import cdd.sqlalchemy.utils.shared_utils
 from cdd.shared.ast_utils import get_value, maybe_type_comment, set_arg, set_value
 from cdd.shared.pure_utils import (
-    PY_GTE_3_9,
     find_module_filepath,
     namespaced_upper_camelcase_to_pascal,
     none_types,
     rpartial,
     tab,
 )
 from cdd.shared.source_transformer import to_code
-from cdd.sqlalchemy.utils.parser_utils import (
+from cdd.sqlalchemy.utils.parse_utils import (
     column_type2typ,
     get_pk_and_type,
     get_table_name,
@@ -76,7 +74,7 @@ def param_to_sqlalchemy_column_call(name_param, include_name):
     x_typ_sql = _param.get("x_typ", {}).get("sql", {})
 
     if "typ" in _param:
-        nullable = update_args_infer_typ_sqlalchemy(
+        nullable = cdd.sqlalchemy.utils.shared_utils.update_args_infer_typ_sqlalchemy(
             _param, args, name, nullable, x_typ_sql
         )
 
@@ -161,145 +159,6 @@ def param_to_sqlalchemy_column_call(name_param, include_name):
     )
 
 
-def update_args_infer_typ_sqlalchemy(_param, args, name, nullable, x_typ_sql):
-    """
-    :param _param: Param with typ
-    :type _param: ```dict```
-
-    :param args:
-    :type args: ```List```
-
-    :param name:
-    :type name: ```str```
-
-    :param nullable:
-    :type nullable: ```Optional[bool]```
-
-    :param x_typ_sql:
-    :type x_typ_sql: ```dict```
-
-    :rtype: ```bool```
-    """
-    if _param["typ"].startswith("Optional["):
-        _param["typ"] = _param["typ"][len("Optional[") : -1]
-        nullable = True
-    if "Literal[" in _param["typ"]:
-        parsed_typ = get_value(ast.parse(_param["typ"]).body[0])
-        assert (
-            parsed_typ.value.id == "Literal"
-        ), "Only basic Literal support is implemented, not {}".format(
-            parsed_typ.value.id
-        )
-        args.append(
-            Call(
-                func=Name("Enum", Load()),
-                args=get_value(parsed_typ.slice).elts,
-                keywords=[
-                    ast.keyword(arg="name", value=set_value(name), identifier=None)
-                ],
-                expr=None,
-                expr_func=None,
-            )
-        )
-    elif _param["typ"].startswith("List["):
-        after_generic = _param["typ"][len("List[") :]
-        if "struct" in after_generic:  # "," in after_generic or
-            name = Name(id="JSON", ctx=Load())
-        else:
-            list_typ = ast.parse(_param["typ"]).body[0]
-            assert isinstance(
-                list_typ, Expr
-            ), "Expected `Expr` got `{type_name}`".format(
-                type_name=type(list_typ).__name__
-            )
-            assert isinstance(
-                list_typ.value, Subscript
-            ), "Expected `Subscript` got `{type_name}`".format(
-                type_name=type(list_typ.value).__name__
-            )
-            name = next(
-                filter(rpartial(isinstance, Name), ast.walk(list_typ.value.slice)), None
-            )
-            assert name is not None, "Could not find a type in {!r}".format(
-                to_code(list_typ.value.slice)
-            )
-        args.append(
-            Call(
-                func=Name(id="ARRAY", ctx=Load()),
-                args=[
-                    Name(
-                        id=typ2column_type.get(name.id, name.id),
-                        ctx=Load(),
-                    )
-                ],
-                keywords=[],
-                expr=None,
-                expr_func=None,
-            )
-        )
-    elif "items" in _param and _param["items"].get("type", False) in typ2column_type:
-        args.append(
-            Call(
-                func=Name(id="ARRAY", ctx=Load()),
-                args=[Name(id=typ2column_type[_param["items"]["type"]], ctx=Load())],
-                keywords=[],
-                expr=None,
-                expr_func=None,
-            )
-        )
-    elif _param.get("typ").startswith("Union["):
-        # Hack to remove the union type. Enum parse seems to be incorrect?
-        union_typ = ast.parse(_param["typ"]).body[0]
-        assert isinstance(
-            union_typ.value, Subscript
-        ), "Expected `Subscript` got `{type_name}`".format(
-            type_name=type(union_typ.value).__name__
-        )
-        union_typ_tuple = (
-            union_typ.value.slice if PY_GTE_3_9 else union_typ.value.slice.value
-        )
-        assert isinstance(
-            union_typ_tuple, Tuple
-        ), "Expected `Tuple` got `{type_name}`".format(
-            type_name=type(union_typ_tuple).__name__
-        )
-        assert (
-            len(union_typ_tuple.elts) == 2
-        ), "Expected length of 2 got `{tuple_len}`".format(
-            tuple_len=len(union_typ_tuple.elts)
-        )
-        left, right = map(attrgetter("id"), union_typ_tuple.elts)
-        args.append(
-            Name(
-                typ2column_type.get(right, right)
-                if left in typ2column_type
-                else typ2column_type.get(left, left),
-                Load(),
-            )
-        )
-    else:
-        type_name = (
-            x_typ_sql["type"]
-            if "type" in x_typ_sql
-            else typ2column_type.get(_param["typ"], _param["typ"])
-        )
-        args.append(
-            Call(
-                func=Name(type_name, Load()),
-                args=list(map(set_value, x_typ_sql.get("type_args", iter(())))),
-                keywords=[
-                    keyword(arg=arg, value=set_value(val))
-                    for arg, val in x_typ_sql.get("type_kwargs", dict()).items()
-                ],
-                expr=None,
-                expr_func=None,
-            )
-            if "type_args" in x_typ_sql or "type_kwargs" in x_typ_sql
-            else Name(type_name, Load())
-        )
-    return nullable
-
-
 def generate_repr_method(params, cls_name, docstring_format):
     """
     Generate a `__repr__` method with all params, using `str.format` syntax
@@ -898,6 +757,8 @@ def sqlalchemy_table_to_class(table_expr_ass):
         "int": "Integer",
         "str": "String",
         "string": "String",
+        "int64": "BigInteger",
+        "Optional[dict]": "JSON",
     }
 )
 
@@ -907,7 +768,7 @@ def sqlalchemy_table_to_class(table_expr_ass):
     "param_to_sqlalchemy_column_call",
     "sqlalchemy_class_to_table",
     "sqlalchemy_table_to_class",
-    "update_args_infer_typ_sqlalchemy",
+    "typ2column_type",
     "update_fk_for_file",
     "update_with_imports_from_columns",
 ]

cdd/json_schema/emit.py

@@ -9,7 +9,7 @@
 
 from cdd.docstring.emit import docstring
 from cdd.json_schema.utils.emit_utils import param2json_schema_property
-from cdd.shared.pure_utils import SetEncoder, deindent, pp
+from cdd.shared.pure_utils import SetEncoder, deindent
 
 
 def json_schema(
@@ -111,7 +111,6 @@ def json_schema_file(input_mapping, output_filename):
     :param output_filename: Output file to write to
     :type output_filename: ```str```
     """
-    pp(dict(input_mapping))
     schemas_it = (json_schema(v) for k, v in input_mapping.items())
     schemas = (
         {"schemas": list(schemas_it)} if len(input_mapping) > 1 else next(schemas_it)

cdd/shared/ast_utils.py

@@ -1644,7 +1644,7 @@ def infer_imports(module):
     :return: Iterable of imports
     :rtype: ```List[Union[Import, ImportFrom]]```
     """
-    import cdd.sqlalchemy.utils.parser_utils  # Should this be a function param instead?
+    import cdd.sqlalchemy.utils.parse_utils  # Should this be a function param instead?
 
     if isinstance(module, (ClassDef, FunctionDef, AsyncFunctionDef, Assign)):
         module = Module(body=[module], type_ignores=[], stmt=None)

cdd/sqlalchemy/emit.py

@@ -9,16 +9,12 @@
 from operator import add
 from os import environ
 
-from cdd.compound.openapi.utils.emit_utils import (
-    ensure_has_primary_key,
-    generate_repr_method,
-    param_to_sqlalchemy_column_call,
-)
+import cdd.compound.openapi.utils.emit_utils
 from cdd.docstring.emit import docstring
 from cdd.shared.ast_utils import maybe_type_comment, set_value
 from cdd.shared.pure_utils import deindent, indent_all_but_first, tab
 
-FORCE_PK_ID = environ.get("FORCE_PK_ID", False) in (False, 0, "0", "false")
+FORCE_PK_ID = environ.get("FORCE_PK_ID", False) not in (False, 0, "0", "false")
 
 
 def sqlalchemy_table(
@@ -87,8 +83,11 @@ def sqlalchemy_table(
                             )
                         ),
                         map(
-                            partial(param_to_sqlalchemy_column_call, include_name=True),
-                            ensure_has_primary_key(
+                            partial(
+                                cdd.compound.openapi.utils.emit_utils.param_to_sqlalchemy_column_call,
+                                include_name=True,
+                            ),
+                            cdd.compound.openapi.utils.emit_utils.ensure_has_primary_key(
                                 intermediate_repr["params"], force_pk_id
                             ).items(),
                         ),
@@ -285,18 +284,18 @@ def _add(a, b):
                     *map(
                         lambda name_param: Assign(
                             targets=[Name(name_param[0], Store())],
-                            value=param_to_sqlalchemy_column_call(
+                            value=cdd.compound.openapi.utils.emit_utils.param_to_sqlalchemy_column_call(
                                 name_param, include_name=False
                             ),
                             expr=None,
                             lineno=None,
                             **maybe_type_comment,
                         ),
-                        ensure_has_primary_key(
+                        cdd.compound.openapi.utils.emit_utils.ensure_has_primary_key(
                             intermediate_repr["params"], force_pk_id
                         ).items(),
                     ),
-                    generate_repr_method(
+                    cdd.compound.openapi.utils.emit_utils.generate_repr_method(
                         intermediate_repr["params"], class_name, docstring_format
                     )
                     if emit_repr

cdd/sqlalchemy/parse.py

@@ -20,7 +20,7 @@
 from cdd.shared.ast_utils import get_value
 from cdd.shared.defaults_utils import extract_default
 from cdd.shared.pure_utils import assert_equal
-from cdd.sqlalchemy.utils.parser_utils import column_call_to_param
+from cdd.sqlalchemy.utils.parse_utils import column_call_to_param
 
 
 def sqlalchemy_table(call_or_name, parse_original_whitespace=False):

cdd/sqlalchemy/utils/parser_utils.py → cdd/sqlalchemy/utils/parse_utils.py

@@ -3,6 +3,7 @@
 """
 
 import ast
+from _ast import Call, Load, Name
 from ast import (
     Assign,
     Call,
@@ -18,6 +19,8 @@
 from itertools import chain, filterfalse
 from operator import attrgetter
 
+from _operator import attrgetter
+
 from cdd.shared.ast_utils import get_value
 from cdd.shared.pure_utils import append_to_dict, rpartial
 from cdd.shared.source_transformer import to_code
@@ -430,7 +433,6 @@ def get_table_name(sqlalchemy_class):
     "str": "str",
 }
 
-
 __all__ = [
     "column_call_name_manipulator",
     "column_call_to_param",