[graphql-pydantic-converter] Fix Map type rendering (#49)

Co-authored-by: Jozef Volak <[email protected]>

Author: Jozefiel

utils/graphql-pydantic-converter/CHANGELOG.md

@@ -10,4 +10,10 @@
 
 # 0.1.2
 - Ignore generating of private graphql schema objects
-- Add graphql_pydantic_converter.graphql_types.concatenate_queries function
+- Add graphql_pydantic_converter.graphql_types.concatenate_queries function
+
+# 1.0.0
+- Migration to pydantic v2
+
+# 1.1.0
+- Change Map type to dict (key, value)

utils/graphql-pydantic-converter/README.md

@@ -44,6 +44,7 @@ options:
 import typing
 
 from pydantic import Field
+from pydantic import PrivateAttr
 from graphql_pydantic_converter.graphql_types import Input
 from graphql_pydantic_converter.graphql_types import Mutation
 from graphql_pydantic_converter.graphql_types import Payload
@@ -58,36 +59,36 @@ String: typing.TypeAlias = str
 
 class CreateScheduleInput(Input):
     name: String
-    workflow_name: String = Field(alias='workflowName')
-    workflow_version: String = Field(alias='workflowVersion')
-    cron_string: String = Field(alias='cronString')
-    enabled: typing.Optional[Boolean]
-    parallel_runs: typing.Optional[Boolean] = Field(alias='parallelRuns')
-    workflow_context: typing.Optional[String] = Field(alias='workflowContext')
-    from_date: typing.Optional[DateTime] = Field(alias='fromDate')
-    to_date: typing.Optional[DateTime] = Field(alias='toDate')
+    workflow_name: String = Field(default=None, alias='workflowName')
+    workflow_version: String = Field(default=None, alias='workflowVersion')
+    cron_string: String = Field(default=None, alias='cronString')
+    enabled: typing.Optional[Boolean] = Field(default=None)
+    parallel_runs: typing.Optional[Boolean] = Field(default=None, alias='parallelRuns')
+    workflow_context: typing.Optional[String] = Field(default=None, alias='workflowContext')
+    from_date: typing.Optional[DateTime] = Field(default=None, alias='fromDate')
+    to_date: typing.Optional[DateTime] = Field(default=None, alias='toDate')
 
 class Schedule(Payload):
-    name: typing.Optional[Boolean] = Field(response='String', default=False)
-    enabled: typing.Optional[Boolean] = Field(response='Boolean', default=False)
-    parallel_runs: typing.Optional[Boolean] = Field(response='Boolean', alias='parallelRuns', default=False)
-    workflow_name: typing.Optional[Boolean] = Field(response='String', alias='workflowName', default=False)
-    workflow_version: typing.Optional[Boolean] = Field(response='String', alias='workflowVersion', default=False)
-    cron_string: typing.Optional[Boolean] = Field(response='String', alias='cronString', default=False)
-    workflow_context: typing.Optional[Boolean] = Field(response='String', alias='workflowContext', default=False)
-    from_date: typing.Optional[Boolean] = Field(response='DateTime', alias='fromDate', default=False)
-    to_date: typing.Optional[Boolean] = Field(response='DateTime', alias='toDate', default=False)
-    status: typing.Optional[Boolean] = Field(response='Status', default=False)
+    name: typing.Optional[bool] = Field(default=False)
+    enabled: typing.Optional[bool] = Field(default=False)
+    parallel_runs: typing.Optional[bool] = Field(alias='parallelRuns', default=False)
+    workflow_name: typing.Optional[bool] = Field(alias='workflowName', default=False)
+    workflow_version: typing.Optional[bool] = Field(alias='workflowVersion', default=False)
+    cron_string: typing.Optional[bool] = Field(alias='cronString', default=False)
+    workflow_context: typing.Optional[bool] = Field(alias='workflowContext', default=False)
+    from_date: typing.Optional[bool] = Field(alias='fromDate', default=False)
+    to_date: typing.Optional[bool] = Field(alias='toDate', default=False)
+    status: typing.Optional[bool] = Field(default=False)
 
 
 class CreateScheduleMutation(Mutation):
-    _name: str = Field('createSchedule', const=True)
+    _name: str = PrivateAttr('createSchedule')
     input: CreateScheduleInput
     payload: Schedule
 
-CreateScheduleInput.update_forward_refs()
-CreateScheduleMutation.update_forward_refs()
-Schedule.update_forward_refs()
+CreateScheduleInput.model_rebuild()
+CreateScheduleMutation.model_rebuild()
+Schedule.model_rebuild()
 
 ```
 
@@ -147,7 +148,7 @@ mutation {
 
 ### Response parser
 
-Example of generated model.py ()
+Example of generated model.py
 
 ```python 
 import typing
@@ -173,25 +174,25 @@ class Status(ENUM):
 
 
 class SchedulePayload(BaseModel):
-    name: typing.Optional[typing.Optional[String]]
-    enabled: typing.Optional[typing.Optional[Boolean]]
-    parallel_runs: typing.Optional[typing.Optional[Boolean]] = Field(alias='parallelRuns')
-    workflow_name: typing.Optional[typing.Optional[String]] = Field(alias='workflowName')
-    workflow_version: typing.Optional[typing.Optional[String]] = Field(alias='workflowVersion')
-    cron_string: typing.Optional[typing.Optional[String]] = Field(alias='cronString')
-    workflow_context: typing.Optional[typing.Optional[String]] = Field(alias='workflowContext')
-    from_date: typing.Optional[typing.Optional[DateTime]] = Field(alias='fromDate')
-    to_date: typing.Optional[typing.Optional[DateTime]] = Field(alias='toDate')
-    status: typing.Optional[typing.Optional[Status]]
+    name: typing.Optional[typing.Optional[String]] = Field(default=None)
+    enabled: typing.Optional[typing.Optional[Boolean]] = Field(default=None)
+    parallel_runs: typing.Optional[typing.Optional[Boolean]] = Field(default=None, alias='parallelRuns')
+    workflow_name: typing.Optional[typing.Optional[String]] = Field(default=None, alias='workflowName')
+    workflow_version: typing.Optional[typing.Optional[String]] = Field(default=None, alias='workflowVersion')
+    cron_string: typing.Optional[typing.Optional[String]] = Field(default=None, alias='cronString')
+    workflow_context: typing.Optional[typing.Optional[String]] = Field(default=None, alias='workflowContext')
+    from_date: typing.Optional[typing.Optional[DateTime]] = Field(default=None, alias='fromDate')
+    to_date: typing.Optional[typing.Optional[DateTime]] = Field(default=None, alias='toDate')
+    status: typing.Optional[typing.Optional[Status]] = Field(default=None)
 
 
 class CreateScheduleData(BaseModel):
-    create_schedule: SchedulePayload = Field(alias='createSchedule')
+    create_schedule: SchedulePayload = Field(default=None, alias='createSchedule')
 
 
 class CreateScheduleResponse(BaseModel):
-    data: typing.Optional[CreateScheduleData]
-    errors: typing.Optional[typing.Any]
+    data: typing.Optional[CreateScheduleData] = Field(default=None)
+    errors: typing.Optional[typing.Any] = Field(default=None)
 
 
 ```

utils/graphql-pydantic-converter/graphql_pydantic_converter/graphql_types.py

@@ -221,21 +221,31 @@ def dict_to_custom_string(self, value: dict[str, Any]) -> str:
             for item in value:
                 pairs.append(self.dict_to_custom_string(item))
             return f"[ {', '.join(pairs)}]"
-        else:
+        elif isinstance(value, dict):
+            pairs = []
+            for key, values in value.items():
+                pairs.append(f'{key}: {self.dict_to_custom_string(values)}')
+            return f'{{ {", ".join(pairs)} }}'
+        elif isinstance(value, str):
             return f'"{value}"'
+        else:
+            return f'{value}'
 
     def render(self) -> str:
-        payload = ''
+        mutation_name: str = self._name
+        payload: str = ''
         if isinstance(self.payload, Payload):
             payload = f'{{ {self.payload.render()} }}'
+
         variables: list[str] = []
-        for k, value in self:
-            if k not in ['_name', 'payload']:
-                variables.append(f' {k}: {self.dict_to_custom_string( value)}')
+        inputs = self.model_dump(exclude={'_name', 'payload'}, exclude_none=True, by_alias=True)
+
+        for name, value in inputs.items():
+            variables.append(f' {name}: {self.dict_to_custom_string(value)}')
+
         variable = ', '.join(variables)
-        name: str = self._name
 
-        return f'mutation {{ { name } ({variable}) {payload} }}'
+        return f'mutation {{ {mutation_name} ({variable}) {payload} }}'
 
 
 class Query(BaseModel):

utils/graphql-pydantic-converter/graphql_pydantic_converter/schema_converter.py

@@ -60,6 +60,7 @@ class Scalars(str, Enum):
         Int = 'Int'
         Boolean = 'Boolean'
         Upload = 'Upload'
+        Map = 'Map'
 
     class QueryType(BaseModel):
         name: Optional[str] = None
@@ -201,14 +202,14 @@ def __import_classes(self, items: dict[str, Any], worker_list: list[Optional[str
 
         imports = [
             'from __future__ import annotations\n\n'
-            'import typing\n',
-            'from pydantic import BaseModel',
-            'from pydantic import Field',
-            'from pydantic import PrivateAttr',
-            '\n'
+            'import typing\n\n',
         ]
         self.__result += '\n'.join(imports)
 
+        self.__result += 'from pydantic import BaseModel\n'
+        self.__result += 'from pydantic import Field\n'
+        self.__result += 'from pydantic import PrivateAttr\n\n'
+
         if items[GraphqlJsonParser.TypeKind.ENUM]:
             self.__result += kv_template.substitute(type=GraphqlJsonParser.ConverterMap.ENUM.value)
         if items[GraphqlJsonParser.TypeKind.INPUT_OBJECT]:
@@ -223,6 +224,7 @@ def __import_classes(self, items: dict[str, Any], worker_list: list[Optional[str
             self.__result += kv_template.substitute(type=GraphqlJsonParser.ConverterMap.QUERY.value)
         if GraphqlJsonParser.ConverterMap.SUBSCRIPTION in worker_list:
             self.__result += kv_template.substitute(type=GraphqlJsonParser.ConverterMap.SUBSCRIPTION.value)
+
         self.__result += '\n'
 
     def __extract_fields(self, of_type: OfType | Type, previous: list[Any]) -> list[Any]:
@@ -305,6 +307,9 @@ def __create_scalar(self, scalars: list[Type]) -> None:
                         scalar_type = 'list'
                     case self.Scalars.ID:
                         scalar_type = 'str'
+                    case self.Scalars.Map:
+                        scalar_type = 'dict[str, typing.Any]'
+
                 self.__result += kv_template.substitute(indent=self.__INDENT, name=scalar.name, type=scalar_type)
                 self.__ignore_enums.append(scalar.name)
 
@@ -360,7 +365,10 @@ def __create_input(self, inputs: list[Type]) -> None:
                         name = field.name
                         field_args = []
                         field_str = ''
-                        if name.startswith('typing.Optional'):
+
+                        kind = self.__build_type(self.__extract_fields(field.type, []))
+
+                        if kind.startswith('typing.Optional'):
                             field_args.append('default=None')
 
                         if self.is_not_snake_case(field.name):
@@ -370,7 +378,6 @@ def __create_input(self, inputs: list[Type]) -> None:
                         if len(field_args) > 0:
                             field_str = f" = Field({', '.join(field_args)})"
 
-                        kind = self.__build_type(self.__extract_fields(field.type, []))
                         self.__result += kv_template.substitute(
                             indent=self.__INDENT, name=name, val=kind, field=field_str
                         )

utils/graphql-pydantic-converter/pyproject.toml

@@ -19,7 +19,7 @@ packages = [{ include = "graphql_pydantic_converter" }]
 name = "graphql-pydantic-converter"
 description = "Convert pydantic schema to pydantic datamodel and build request from it"
 authors = ["Jozef Volak <[email protected]>"]
-version = '1.0.0'
+version = '1.0.1'
 readme = ["README.md", "CHANGELOG.md"]
 keywords = ["graphql", "pydantic"]
 license = "Apache 2.0"
@@ -79,3 +79,6 @@ init_forbid_extra = true
 init_typed = true
 warn_required_dynamic_aliases = true
 warn_untyped_fields = true
+
+[tool.pytest.ini_options]
+python_files = 'tests/*'

utils/graphql-pydantic-converter/tests/model.py

@@ -56,20 +56,20 @@ class AddBlueprintInput(Input):
 
 
 class AddDeviceInput(Input):
-    address: typing.Optional[String]
-    blueprint_id: typing.Optional[String] = Field(alias='blueprintId')
-    device_size: typing.Optional[DeviceSize] = Field(alias='deviceSize')
-    device_type: typing.Optional[String] = Field(alias='deviceType')
-    label_ids: typing.Optional[list[String]] = Field(alias='labelIds')
-    model: typing.Optional[String]
-    mount_parameters: typing.Optional[String] = Field(alias='mountParameters')
+    address: typing.Optional[String] = Field(default=None)
+    blueprint_id: typing.Optional[String] = Field(default=None, alias='blueprintId')
+    device_size: typing.Optional[DeviceSize] = Field(default=None, alias='deviceSize')
+    device_type: typing.Optional[String] = Field(default=None, alias='deviceType')
+    label_ids: typing.Optional[list[String]] = Field(default=None, alias='labelIds')
+    model: typing.Optional[String] = Field(default=None)
+    mount_parameters: typing.Optional[String] = Field(default=None, alias='mountParameters')
     name: String
-    password: typing.Optional[String]
-    port: typing.Optional[Int]
-    service_state: typing.Optional[DeviceServiceState] = Field(alias='serviceState')
-    username: typing.Optional[String]
-    vendor: typing.Optional[String]
-    version: typing.Optional[String]
+    password: typing.Optional[String] = Field(default=None)
+    port: typing.Optional[Int] = Field(default=None)
+    service_state: typing.Optional[DeviceServiceState] = Field(default=None, alias='serviceState')
+    username: typing.Optional[String] = Field(default=None)
+    vendor: typing.Optional[String] = Field(default=None)
+    version: typing.Optional[String] = Field(default=None)
     zone_id: String = Field(alias='zoneId')
 
 
@@ -80,8 +80,8 @@ class GraphNodeCoordinatesInput(Input):
 
 
 class UpdateBlueprintInput(Input):
-    name: typing.Optional[String]
-    template: typing.Optional[String]
+    name: typing.Optional[String] = Field(default=None)
+    template: typing.Optional[String] = Field(default=None)
 
 
 class AddBlueprintPayload(Payload):

utils/graphql-pydantic-converter/tests/render_models.py

@@ -4,9 +4,18 @@
 from pydantic import PrivateAttr
 
 from graphql_pydantic_converter.graphql_types import Input
+from graphql_pydantic_converter.graphql_types import Mutation
 from graphql_pydantic_converter.graphql_types import Payload
 from graphql_pydantic_converter.graphql_types import Query
 
+Boolean: typing.TypeAlias = bool
+Cursor: typing.TypeAlias = typing.Any
+Float: typing.TypeAlias = float
+ID: typing.TypeAlias = str
+Int: typing.TypeAlias = int
+Map: typing.TypeAlias = dict[str, typing.Any]
+String: typing.TypeAlias = str
+
 
 class TagAnd(Input):
     matches_all: typing.Optional[list[str]] = Field(default=None, alias='matchesAll')
@@ -109,3 +118,20 @@ class SchedulesQuery(Query):
     last: typing.Optional[int] = Field(default=None)
     filter: typing.Optional[SchedulesFilterInput] = Field(default=None)
     payload: ScheduleConnection
+
+
+class Resource(Payload):
+    description: typing.Optional[Boolean] = Field(default=False, alias='Description')
+    nested_pool: typing.Optional[ResourcePool] = Field(default=None, alias='NestedPool')
+    parent_pool: typing.Optional[ResourcePool] = Field(default=None, alias='ParentPool')
+    properties: typing.Optional[Boolean] = Field(default=False, alias='Properties')
+    alternative_id: typing.Optional[Boolean] = Field(default=False, alias='AlternativeId')
+    id: typing.Optional[Boolean] = Field(default=False)
+
+
+class ClaimResourceMutation(Mutation):
+    _name: str = PrivateAttr('ClaimResource')
+    pool_id: ID = Field(alias='poolId')
+    description: typing.Optional[String] = Field(default=None)
+    user_input: Map = Field(alias='userInput')
+    payload: Resource

utils/graphql-pydantic-converter/tests/test_graphql_generator.py

@@ -10,7 +10,9 @@
 from model import BlueprintsQuery
 from pydantic import Field
 from render_models import AllocationStrategy
+from render_models import ClaimResourceMutation
 from render_models import PageInfoSchedule
+from render_models import Resource
 from render_models import ResourcePool
 from render_models import ResourcePoolConnection
 from render_models import ResourcePoolEdge
@@ -73,6 +75,21 @@ def test_render_mutation(self) -> None:
         ).render()
         assert reference == mutation
 
+        reference = ('mutation { ClaimResource ( poolId: "00000000001",  description: "<description>",  userInput: '
+                     '{ address: "0.0.0.0", port: 80 }) { Properties AlternativeId id } }')
+        mutation = ClaimResourceMutation(
+            poolId='00000000001',
+            description='<description>',
+            userInput={'address': '0.0.0.0', 'port': 80},
+            payload=Resource(
+                id=True,
+                Properties=True,
+                AlternativeId=True
+            )
+        ).render()
+
+        assert reference == mutation
+
     def test_render_input(self) -> None:
         class DeviceSize(ENUM):
             SMALL = 'SMALL'
@@ -274,4 +291,3 @@ def test_multiple_query(self) -> None:
 
         merged_query = graphql_pydantic_converter.graphql_types.concatenate_queries(queries)
         assert reference == merged_query
-