"""
GraphQL Builder Module
This module provides fluent builders for constructing GraphQL queries and mutations.
Example:
# Mutation
request = (
MutationBuilder("updateLead")
.variable("lead", "UpdateLeadInput!", required=True)
.operation("updateLead", args={"lead": "$lead"})
.select("leadId", "name")
.build()
)
# Query
request = (
QueryBuilder("getPolicies")
.variable("ids", "[Int!]!", required=True)
.operation("policies", args={"ids": "$ids"})
.select("policyId", "policyNumber")
.build()
)
"""
from typing import Dict, Any, List, Optional, Union
from ace.core.graphql.request import GraphQLRequest
from ace.core.graphql.selection import SelectionBuilder, SelectionSet
[docs]
class VariableDefinition:
"""Represents a GraphQL variable definition."""
[docs]
def __init__(self, name: str, graphql_type: str, required: bool = False, default: Any = None):
self.name = name
self.graphql_type = graphql_type
self.required = required
self.default = default
[docs]
def build(self) -> str:
"""Build the variable definition string."""
type_str = self.graphql_type
# Add ! if required and not already present
if self.required and not type_str.endswith("!"):
type_str += "!"
result = f"${self.name}: {type_str}"
if self.default is not None:
if isinstance(self.default, str):
result += f' = "{self.default}"'
elif isinstance(self.default, bool):
result += f" = {str(self.default).lower()}"
else:
result += f" = {self.default}"
return result
[docs]
class OperationBuilder(SelectionBuilder):
"""
Builder for a GraphQL operation (the root field of a query/mutation).
Extends SelectionBuilder to provide the same fluent API for field selection.
"""
[docs]
def __init__(
self,
operation_type: str,
operation_name: Optional[str] = None
):
"""
Initialize an OperationBuilder.
Args:
operation_type: Either 'query' or 'mutation'
operation_name: Name of the operation (optional but recommended)
"""
super().__init__(parent=None)
self._operation_type = operation_type
self._operation_name = operation_name
self._variables: List[VariableDefinition] = []
self._root_field: Optional[str] = None
self._root_args: Dict[str, Any] = {}
[docs]
def variable(
self,
name: str,
graphql_type: str,
required: bool = False,
default: Any = None
) -> "OperationBuilder":
"""
Add a variable definition.
Args:
name: Variable name (without $)
graphql_type: GraphQL type (e.g., "Int!", "[String!]!", "UpdateLeadInput!")
required: Whether the variable is required (adds ! if not in type)
default: Default value for the variable
Returns:
Self for chaining
"""
var_def = VariableDefinition(name, graphql_type, required, default)
self._variables.append(var_def)
return self
[docs]
def operation(self, field_name: str, args: Optional[Dict[str, Any]] = None) -> "OperationBuilder":
"""
Set the root operation field.
Args:
field_name: Name of the root field (e.g., "updateLead", "policies")
args: Arguments for the root field
Returns:
Self for chaining
"""
self._root_field = field_name
self._root_args = args or {}
return self
def _build_variables_definition(self) -> str:
"""Build the variables definition section."""
if not self._variables:
return ""
var_strs = [v.build() for v in self._variables]
return f"({', '.join(var_strs)})"
def _build_root_args(self) -> str:
"""Build the root operation arguments."""
if not self._root_args:
return ""
args_parts = []
for key, value in self._root_args.items():
if isinstance(value, str) and value.startswith("$"):
args_parts.append(f"{key}: {value}")
elif isinstance(value, str):
args_parts.append(f'{key}: "{value}"')
elif isinstance(value, bool):
args_parts.append(f"{key}: {str(value).lower()}")
elif value is None:
args_parts.append(f"{key}: null")
else:
args_parts.append(f"{key}: {value}")
return f"({', '.join(args_parts)})"
[docs]
def build(self) -> GraphQLRequest:
"""
Build the complete GraphQL request.
Returns:
GraphQLRequest ready for execution
"""
if not self._root_field:
raise ValueError("Operation field not set. Call .operation() before .build()")
# Build operation header
vars_def = self._build_variables_definition()
op_name = self._operation_name or ""
if op_name and vars_def:
header = f"{self._operation_type} {op_name}{vars_def}"
elif op_name:
header = f"{self._operation_type} {op_name}"
elif vars_def:
header = f"{self._operation_type}{vars_def}"
else:
header = self._operation_type
# Build root field with args
root_args = self._build_root_args()
selection = super().build()
if selection:
query = f"{header} {{ {self._root_field}{root_args} {{ {selection} }} }}"
else:
query = f"{header} {{ {self._root_field}{root_args} }}"
return GraphQLRequest(
query=query,
operation_name=self._operation_name,
operation_type=self._operation_type
)
[docs]
def end(self) -> "OperationBuilder":
"""Override to always return self at root level."""
return self
[docs]
class MutationBuilder(OperationBuilder):
"""
Builder specifically for GraphQL mutations.
Example:
request = (
MutationBuilder("createPolicy")
.variable("policy", "PolicyCreateParamsInput!", required=True)
.variable("participants", "[ParticipantEntityLinkCreateParamsInput!]!", required=True)
.operation("createPolicy", args={"policy": "$policy"})
.select("policyId")
.nested("info")
.select("policyId", "policyNumber", "policyGUID")
.nested("coverages")
.select("id", "coverageNumber")
.end()
.end()
.nested("linkParticipants", args={"linkParticipants": "$participants"})
.select("id")
.nested("party").select("id").end()
.end()
.build()
)
"""
[docs]
def __init__(self, name: Optional[str] = None):
"""
Initialize a MutationBuilder.
Args:
name: Name of the mutation operation (optional but recommended)
"""
super().__init__(operation_type="mutation", operation_name=name)
[docs]
class QueryBuilder(OperationBuilder):
"""
Builder specifically for GraphQL queries.
Example:
request = (
QueryBuilder("getPolicies")
.variable("ids", "[Int!]!", required=True)
.operation("policies", args={"ids": "$ids"})
.select("policyId", "policyNumber")
.nested("product")
.select("productTypeId")
.nested("carrierName").select("default").end()
.end()
.build()
)
"""
[docs]
def __init__(self, name: Optional[str] = None):
"""
Initialize a QueryBuilder.
Args:
name: Name of the query operation (optional but recommended)
"""
super().__init__(operation_type="query", operation_name=name)