diff --git a/example/api/src/queries/hello.py b/example/api/src/queries/hello.py index eb512dbb..58c747b1 100644 --- a/example/api/src/queries/hello.py +++ b/example/api/src/queries/hello.py @@ -1,5 +1,4 @@ -import graphene - +from cpl.api.middleware.request import get_request from cpl.graphql.schema.query import Query @@ -8,6 +7,5 @@ class HelloQuery(Query): Query.__init__(self) self.string_field( "message", - args={"name": graphene.String(default_value="world")}, - resolver=lambda *_, name: f"Hello {name}", - ) + resolver=lambda *_, name: f"Hello {name} {get_request().state.request_id}", + ).with_argument(str, "name", "Name to greet", "world") diff --git a/src/cpl-graphql/cpl/graphql/schema/argument.py b/src/cpl-graphql/cpl/graphql/schema/argument.py new file mode 100644 index 00000000..2f3b938c --- /dev/null +++ b/src/cpl-graphql/cpl/graphql/schema/argument.py @@ -0,0 +1,22 @@ +class Argument: + def __init__(self, t: type, name: str, description: str = None, default_value=None): + self._type = t + self._name = name + self._description = description + self._default_value = default_value + + @property + def type(self) -> type: + return self._type + + @property + def name(self) -> str: + return self._name + + @property + def description(self) -> str | None: + return self._description + + @property + def default_value(self): + return self._default_value diff --git a/src/cpl-graphql/cpl/graphql/schema/field.py b/src/cpl-graphql/cpl/graphql/schema/field.py index 13261675..2744f6d2 100644 --- a/src/cpl-graphql/cpl/graphql/schema/field.py +++ b/src/cpl-graphql/cpl/graphql/schema/field.py @@ -1,20 +1,25 @@ -from cpl.graphql.schema.query import Query +from typing import Self + +from cpl.graphql.schema.argument import Argument +from cpl.graphql.typing import TQuery class Field: - def __init__(self, name: str, gql_type: str, resolver: callable, args: dict | None = None, subquery: Query | None = None): + + def __init__(self, name: str, gql_type: type, resolver: callable, subquery: TQuery | None = None): self._name = name self._gql_type = gql_type self._resolver = resolver - self._args = args or {} - self._subquery: Query | None = subquery + self._subquery = subquery + + self._args: dict[str, Argument] = {} @property def name(self) -> str: return self._name @property - def type(self) -> str: + def type(self) -> type: return self._gql_type @property @@ -26,5 +31,19 @@ class Field: return self._args @property - def subquery(self) -> Query | None: - return self._subquery \ No newline at end of file + def subquery(self) -> TQuery | None: + return self._subquery + + def with_argument(self, arg_type: type, name: str, description: str = None, default_value=None) -> Self: + if name in self._args: + raise ValueError(f"Argument with name '{name}' already exists in field '{self._name}'") + self._args[name] = Argument(name, arg_type, description, default_value) + return self + + def with_arguments(self, args: list[Argument]) -> Self: + for arg in args: + if not isinstance(arg, Argument): + raise ValueError(f"Expected Argument instance, got {type(arg)}") + + self.with_argument(arg.name, arg.type, arg.description, arg.default_value) + return self diff --git a/src/cpl-graphql/cpl/graphql/schema/query.py b/src/cpl-graphql/cpl/graphql/schema/query.py index 13f4c62f..ba93c3a1 100644 --- a/src/cpl-graphql/cpl/graphql/schema/query.py +++ b/src/cpl-graphql/cpl/graphql/schema/query.py @@ -1,7 +1,11 @@ -from typing import Callable, Any, Type +from typing import Callable, Type from graphene import ObjectType +from cpl.graphql.schema.field import Field +from cpl.graphql.typing import Resolver +from cpl.graphql.utils.type_converter import TypeConverter + class Query(ObjectType): @@ -11,44 +15,35 @@ class Query(ObjectType): ObjectType.__init__(self) self._fields: dict[str, Field] = {} - def get_fields(self) -> dict[str, "Field"]: + def get_fields(self) -> dict[str, Field]: return self._fields def field( - self, - name: str, - t: type, - args: dict[str, Any] | None = None, - resolver: Callable | None = None, - ): - gql_type_map: dict[object, str] = { - str: "String", - int: "Int", - float: "Float", - bool: "Boolean", - } - - if t not in gql_type_map: - raise ValueError(f"Unsupported field type: {t}") - + self, + name: str, + t: type, + resolver: Callable | None = None, + ) -> "Field": from cpl.graphql.schema.field import Field - self._fields[name] = Field(name, "String", resolver, args) + self._fields[name] = Field(name, t, resolver) + return self._fields[name] def with_query(self, name: str, subquery: Type["Query"]): from cpl.graphql.schema.field import Field - f = Field(name=name, gql_type="Object", resolver=lambda root, info, **kwargs: {}, subquery=subquery) + f = Field(name=name, gql_type=object, resolver=lambda root, info, **kwargs: {}, subquery=subquery) self._fields[name] = f + return self._fields[name] - def string_field(self, name: str, args: dict[str, Any] | None = None, resolver: Callable | None = None): - self.field(name, str, args, resolver) + def string_field(self, name: str, resolver: Resolver = None) -> "Field": + return self.field(name, str, resolver) - def int_field(self, name: str, args: dict[str, Any] | None = None, resolver: Callable | None = None): - self.field(name, int, args, resolver) + def int_field(self, name: str, resolver: Resolver = None) -> "Field": + return self.field(name, int, resolver) - def float_field(self, name: str, args: dict[str, Any] | None = None, resolver: Callable | None = None): - self.field(name, float, args, resolver) + def float_field(self, name: str, resolver: Resolver = None) -> "Field": + return self.field(name, float, resolver) - def bool_field(self, name: str, args: dict[str, Any] | None = None, resolver: Callable | None = None): - self.field(name, bool, args, resolver) + def bool_field(self, name: str, resolver: Resolver = None) -> "Field": + return self.field(name, bool, resolver) diff --git a/src/cpl-graphql/cpl/graphql/service/schema.py b/src/cpl-graphql/cpl/graphql/service/schema.py index 0dcf02b6..69e76e48 100644 --- a/src/cpl-graphql/cpl/graphql/service/schema.py +++ b/src/cpl-graphql/cpl/graphql/service/schema.py @@ -1,9 +1,14 @@ +from typing import Type + import graphene from cpl.api.logger import APILogger from cpl.dependency.service_provider import ServiceProvider +from cpl.graphql.schema.argument import Argument from cpl.graphql.schema.query import Query from cpl.graphql.schema.root_query import RootQuery +from cpl.graphql.typing import Resolver +from cpl.graphql.utils.type_converter import TypeConverter class Schema: @@ -31,26 +36,29 @@ class Schema: ) return self._schema + @staticmethod + def _field_to_graphene(t: Type[graphene.Scalar] | type, args: dict[str, Argument] = None, resolver: Resolver = None) -> graphene.Field: + arguments = {} + if args is not None: + arguments = { + arg.name: graphene.Argument(TypeConverter.to_graphene(arg.type), description=arg.description, default_value=arg.default_value) + for arg in args.values() + } + + return graphene.Field(t, args=arguments, resolver=resolver) + def to_graphene(self, query: Query, name: str | None = None): assert query is not None, "Query cannot be None" attrs = {} for field in query.get_fields().values(): - if field.type == "String": - attrs[field.name] = graphene.Field( - graphene.String, - **field.args, - resolver=field.resolver - ) - - elif field.type == "Object" and field.subquery is not None: + if field.type == object and field.subquery is not None: subquery = self._provider.get_service(field.subquery) sub = self.to_graphene(subquery, name=field.name.capitalize()) - attrs[field.name] = graphene.Field( - sub, - **field.args, - resolver=field.resolver - ) + attrs[field.name] = self._field_to_graphene(sub, field.args, field.resolver) + continue + + attrs[field.name] = self._field_to_graphene(TypeConverter.to_graphene(field.type), field.args, field.resolver) class_name = name or query.__class__.__name__ return type(class_name, (graphene.ObjectType,), attrs) diff --git a/src/cpl-graphql/cpl/graphql/typing.py b/src/cpl-graphql/cpl/graphql/typing.py index 58587f3f..d5b63494 100644 --- a/src/cpl-graphql/cpl/graphql/typing.py +++ b/src/cpl-graphql/cpl/graphql/typing.py @@ -1,5 +1,5 @@ -from typing import Type +from typing import Type, Callable -from cpl.graphql.schema.query import Query - -TQuery = Type[Query] \ No newline at end of file +TQuery = Type["Query"] +Resolver = Callable +ScalarType = str | int | float | bool | object \ No newline at end of file diff --git a/src/cpl-graphql/cpl/graphql/utils/__init__.py b/src/cpl-graphql/cpl/graphql/utils/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/cpl-graphql/cpl/graphql/utils/type_converter.py b/src/cpl-graphql/cpl/graphql/utils/type_converter.py new file mode 100644 index 00000000..c89b8928 --- /dev/null +++ b/src/cpl-graphql/cpl/graphql/utils/type_converter.py @@ -0,0 +1,38 @@ +from typing import Type + +import graphene + +from cpl.graphql.typing import ScalarType + + +class TypeConverter: + + @staticmethod + def from_graphene(t: Type[graphene.Scalar]) -> ScalarType: + graphene_type_map: dict[Type[graphene.Scalar], ScalarType] = { + graphene.String: str, + graphene.Int: int, + graphene.Float: float, + graphene.Boolean: bool, + graphene.ObjectType: object, + } + + if t not in graphene_type_map: + raise ValueError(f"Unsupported field type: {t}") + + return graphene_type_map[t] + + @staticmethod + def to_graphene(t: ScalarType) -> Type[graphene.Scalar]: + type_graphene_map: dict[ScalarType, Type[graphene.Scalar]] = { + str: graphene.String, + int: graphene.Int, + float: graphene.Float, + bool: graphene.Boolean, + object: graphene.ObjectType, + } + + if t not in type_graphene_map: + raise ValueError(f"Unsupported field type: {t}") + + return type_graphene_map[t]