WIP: dev into master #184
@@ -1,54 +0,0 @@
|
||||
from typing import Callable, Any, Type
|
||||
|
||||
from graphene import ObjectType
|
||||
|
||||
|
||||
class QueryBase(ObjectType):
|
||||
|
||||
def __init__(self):
|
||||
from cpl.graphql.schema.field import Field
|
||||
|
||||
ObjectType.__init__(self)
|
||||
self._fields: 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}")
|
||||
|
||||
from cpl.graphql.schema.field import Field
|
||||
|
||||
self._fields[name] = Field(name, "String", resolver, args)
|
||||
|
||||
def with_query(self, name: str, subquery: Type["QueryBase"]):
|
||||
from cpl.graphql.schema.field import Field
|
||||
|
||||
f = Field(name=name, gql_type="Object", resolver=lambda root, info, **kwargs: {}, subquery=subquery)
|
||||
self._fields[name] = f
|
||||
|
||||
def string_field(self, name: str, args: dict[str, Any] | None = None, resolver: Callable | None = None):
|
||||
self.field(name, str, args, resolver)
|
||||
|
||||
def int_field(self, name: str, args: dict[str, Any] | None = None, resolver: Callable | None = None):
|
||||
self.field(name, int, args, resolver)
|
||||
|
||||
def float_field(self, name: str, args: dict[str, Any] | None = None, resolver: Callable | None = None):
|
||||
self.field(name, float, args, resolver)
|
||||
|
||||
def bool_field(self, name: str, args: dict[str, Any] | None = None, resolver: Callable | None = None):
|
||||
self.field(name, bool, args, resolver)
|
||||
@@ -1,4 +1,4 @@
|
||||
from cpl.api import ApiModule
|
||||
from cpl.api.api_module import ApiModule
|
||||
from cpl.dependency.module.module import Module
|
||||
from cpl.dependency.service_provider import ServiceProvider
|
||||
from cpl.graphql.schema.root_query import RootQuery
|
||||
@@ -8,8 +8,8 @@ from cpl.graphql.service.service import GraphQLService
|
||||
|
||||
class GraphQLModule(Module):
|
||||
dependencies = [ApiModule]
|
||||
singleton = [Schema, RootQuery]
|
||||
scoped = [GraphQLService]
|
||||
singleton = [Schema]
|
||||
scoped = [GraphQLService, RootQuery]
|
||||
|
||||
@staticmethod
|
||||
def configure(services: ServiceProvider) -> None:
|
||||
|
||||
@@ -1,13 +1,13 @@
|
||||
from cpl.graphql.abc.query_base import QueryBase
|
||||
from cpl.graphql.schema.query import Query
|
||||
|
||||
|
||||
class Field:
|
||||
def __init__(self, name: str, gql_type: str, resolver: callable, args: dict | None = None, subquery: QueryBase | None = None):
|
||||
def __init__(self, name: str, gql_type: str, resolver: callable, args: dict | None = None, subquery: Query | None = None):
|
||||
self._name = name
|
||||
self._gql_type = gql_type
|
||||
self._resolver = resolver
|
||||
self._args = args or {}
|
||||
self._subquery: QueryBase | None = subquery
|
||||
self._subquery: Query | None = subquery
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
@@ -26,5 +26,5 @@ class Field:
|
||||
return self._args
|
||||
|
||||
@property
|
||||
def subquery(self) -> QueryBase | None:
|
||||
def subquery(self) -> Query | None:
|
||||
return self._subquery
|
||||
@@ -1,6 +1,54 @@
|
||||
from cpl.graphql.abc.query_base import QueryBase
|
||||
from typing import Callable, Any, Type
|
||||
|
||||
from graphene import ObjectType
|
||||
|
||||
|
||||
class Query(QueryBase):
|
||||
class Query(ObjectType):
|
||||
|
||||
def __init__(self):
|
||||
QueryBase.__init__(self)
|
||||
from cpl.graphql.schema.field import Field
|
||||
|
||||
ObjectType.__init__(self)
|
||||
self._fields: 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}")
|
||||
|
||||
from cpl.graphql.schema.field import Field
|
||||
|
||||
self._fields[name] = Field(name, "String", resolver, args)
|
||||
|
||||
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)
|
||||
self._fields[name] = f
|
||||
|
||||
def string_field(self, name: str, args: dict[str, Any] | None = None, resolver: Callable | None = None):
|
||||
self.field(name, str, args, resolver)
|
||||
|
||||
def int_field(self, name: str, args: dict[str, Any] | None = None, resolver: Callable | None = None):
|
||||
self.field(name, int, args, resolver)
|
||||
|
||||
def float_field(self, name: str, args: dict[str, Any] | None = None, resolver: Callable | None = None):
|
||||
self.field(name, float, args, resolver)
|
||||
|
||||
def bool_field(self, name: str, args: dict[str, Any] | None = None, resolver: Callable | None = None):
|
||||
self.field(name, bool, args, resolver)
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import graphene
|
||||
|
||||
from cpl.api import APILogger
|
||||
from cpl.api.logger import APILogger
|
||||
from cpl.dependency.service_provider import ServiceProvider
|
||||
from cpl.graphql.schema.query import Query
|
||||
from cpl.graphql.schema.root_query import RootQuery
|
||||
|
||||
@@ -6,9 +6,9 @@ from cpl.graphql.service.schema import Schema
|
||||
|
||||
class GraphQLService:
|
||||
def __init__(self, schema: Schema):
|
||||
self._schema = schema.schema
|
||||
if self._schema is None:
|
||||
if schema.schema is None:
|
||||
raise ValueError("Schema has not been built. Call schema.build() before using the service.")
|
||||
self._schema = schema.schema
|
||||
|
||||
async def execute(
|
||||
self,
|
||||
|
||||
Reference in New Issue
Block a user