2nd PoC for Typed GQL Queries Design Doc

Makefile

@@ -77,6 +77,12 @@ clean:
 	@find . -name "__pycache__" -type d -print0 | xargs -0 rm -rf
 	@find . -name "*.pyc" -delete
 
+generate-queries:
+	. ./venv/bin/activate && query-generator
+
+get-introspection:
+	. ./venv/bin/activate && python3 code_generator/introspection.py
+
 dev-venv: clean ## Create a local venv for your IDE and remote debugging
 	python3.9 -m venv venv
 	. ./venv/bin/activate && pip install --upgrade pip
@@ -87,4 +93,4 @@ print-files-modified-in-last-30-days:
 	@git log --since '$(shell date --date='-30 day' +"%m/%d/%y")' --until '$(shell date +"%m/%d/%y")' --oneline --name-only --pretty=format: | sort | uniq | grep -E '.py$$'
 
 format:
-	@. ./venv/bin/activate && black reconcile/ tools/ e2e_tests/
+	@. ./venv/bin/activate && black reconcile/ tools/ e2e_tests/ code_generator/

code_generator/introspection.py

@@ -0,0 +1,21 @@
+import json
+import requests
+
+from graphql import get_introspection_query
+
+
+INTROSPECTION_JSON_FILE = "reconcile/gql_queries/schema_introspection.json"
+
+
+def query_schema():
+    gql_url = "http://localhost:4000/graphql"
+    query = get_introspection_query()
+    request = requests.post(gql_url, json={"query": query})
+    if request.status_code == 200:
+        with open(INTROSPECTION_JSON_FILE, "w") as f:
+            f.write(json.dumps(request.json(), indent=4))
+            return
+    raise Exception(f"Could not query {gql_url}")
+
+
+query_schema()

code_generator/main.py

@@ -0,0 +1,68 @@
+import json
+import os
+
+from graphql import build_client_schema
+from code_generator.introspection import INTROSPECTION_JSON_FILE
+
+from code_generator.query_parser import ParsedNode, ParsedInlineFragmentNode, QueryParser  # type: ignore
+
+
+HEADER = '"""\nTHIS IS AN AUTO-GENERATED FILE. DO NOT MODIFY MANUALLY!\n"""\n'
+
+
+def get_schema() -> dict:
+    with open(INTROSPECTION_JSON_FILE) as f:
+        return json.loads(f.read())["data"]
+
+
+def find_query_files() -> list[str]:
+    result: list[str] = []
+    for root, _, files in os.walk("reconcile/gql_queries"):
+        for name in files:
+            if name.endswith(".gql"):
+                result.append(os.path.join(root, name))
+    return result
+
+
+def traverse(node: ParsedNode) -> str:
+    """
+    Pydantic doesnt play well with from __future__ import annotations
+    --> order of class declaration is important:
+    - post-order for non-inline fragment nodes, i.e., non-interface nodes
+    - pre-order for nodes that implement an interface
+    """
+    result = ""
+    for child in node.fields:
+        if not isinstance(child, ParsedInlineFragmentNode):
+            result = f"{result}{traverse(child)}"
+
+    result = f"{result}{node.class_code_string()}"
+
+    for child in node.fields:
+        if isinstance(child, ParsedInlineFragmentNode):
+            result = f"{result}{traverse(child)}"
+    return result
+
+
+def main():
+    schema_raw = get_schema()
+    schema = build_client_schema(schema_raw)  # type: ignore
+    query_parser = QueryParser(schema=schema)
+    for file in find_query_files():
+        with open(file, "r") as f:
+            result = query_parser.parse(f.read())
+            code = traverse(result)
+            with open(f"{file[:-3]}py", "w") as out_file:
+                out_file.write(HEADER)
+                # TODO: import DateTime
+                out_file.write(
+                    "from typing import Optional, Union  # noqa: F401 # pylint: disable=W0611\n\n"
+                )
+                out_file.write(
+                    "from pydantic import BaseModel, Extra, Field, Json  # noqa: F401  # pylint: disable=W0611"
+                )
+                out_file.write(code)
+                out_file.write("\n")
+
+
+main()

code_generator/mapper.py

@@ -0,0 +1,30 @@
+import re
+
+from graphql import GraphQLOutputType
+
+
+def _keyword_sanitizer(s: str) -> str:
+    if s in ("global", "from", "type", "id", "to", "format"):
+        return f"f_{s}"
+    return s
+
+
+def graphql_primitive_to_python(graphql_type: GraphQLOutputType) -> str:
+    mapping = {
+        "ID": "str",
+        "String": "str",
+        "Int": "int",
+        "Float": "float",
+        "Boolean": "bool",
+        "DateTime": "DateTime",
+        "JSON": "Json",
+    }
+    return mapping.get(str(graphql_type), str(graphql_type))
+
+
+def graphql_field_name_to_python(name: str) -> str:
+    parts = re.split("(?=[A-Z])", name)
+    for i, el in enumerate(parts):
+        parts[i] = el.lower()
+
+    return _keyword_sanitizer("_".join(parts))