Added possibility to put auth schema to root graphs #181
Some checks failed
Test before pr merge / test-lint (pull_request) Failing after 6s
Some checks failed
Test before pr merge / test-lint (pull_request) Failing after 6s
This commit is contained in:
@@ -96,6 +96,9 @@ def main():
|
||||
|
||||
schema.mutation.with_mutation("post", PostMutation).with_public()
|
||||
|
||||
app.with_auth_root_queries(True)
|
||||
app.with_auth_root_mutations(True)
|
||||
|
||||
app.with_playground()
|
||||
app.with_graphiql()
|
||||
|
||||
|
||||
@@ -5,11 +5,16 @@ from cpl.api.application import WebApp
|
||||
from cpl.api.model.validation_match import ValidationMatch
|
||||
from cpl.dependency.service_provider import ServiceProvider
|
||||
from cpl.dependency.typing import Modules
|
||||
from queries.user import UserGraphType, UserFilter, UserSort
|
||||
from .._endpoints.graphiql import graphiql_endpoint
|
||||
from .._endpoints.graphql import graphql_endpoint
|
||||
from .._endpoints.playground import playground_endpoint
|
||||
from ..auth.administration.user.user_mutation import UserMutation
|
||||
from ..graphql_module import GraphQLModule
|
||||
from ..service.schema import Schema
|
||||
from ...application.abc.application_abc import __not_implemented__
|
||||
from ...auth.schema import UserDao
|
||||
from ...core.configuration import Configuration
|
||||
|
||||
|
||||
class GraphQLApp(WebApp):
|
||||
@@ -84,6 +89,19 @@ class GraphQLApp(WebApp):
|
||||
self._with_playground = True
|
||||
return self
|
||||
|
||||
def with_auth_root_queries(self, public: bool = False):
|
||||
if not Configuration.get("GraphQLAuthModuleEnabled", False):
|
||||
raise Exception("GraphQLAuthModule is not loaded yet. Make sure to run 'add_module(GraphQLAuthModule)'")
|
||||
|
||||
schema = self._services.get_service(Schema)
|
||||
schema.query.dao_collection_field(UserGraphType, UserDao, "users", UserFilter, UserSort).with_public(public)
|
||||
|
||||
def with_auth_root_mutations(self, public: bool = False):
|
||||
if not Configuration.get("GraphQLAuthModuleEnabled", False):
|
||||
raise Exception("GraphQLAuthModule is not loaded yet. Make sure to run 'add_module(GraphQLAuthModule)'")
|
||||
|
||||
schema = self._services.get_service(Schema)
|
||||
schema.mutation.with_mutation("user", UserMutation).with_public(public)
|
||||
|
||||
async def _log_before_startup(self):
|
||||
self._logger.info(f"Start API on {self._api_settings.host}:{self._api_settings.port}")
|
||||
|
||||
@@ -21,3 +21,5 @@ class GraphQLAuthModule(Module):
|
||||
def configure(provider: ServiceProvider):
|
||||
schema = provider.get_service(Schema)
|
||||
schema.with_type(UserGraphType)
|
||||
|
||||
|
||||
|
||||
@@ -7,13 +7,15 @@ from cpl.dependency import get_provider
|
||||
from cpl.graphql.abc.strawberry_protocol import StrawberryProtocol
|
||||
|
||||
|
||||
class CollectionGraphTypeFactory:
|
||||
_cache: Dict[Type, Type] = {}
|
||||
from cpl.graphql.utils.type_collector import TypeCollector
|
||||
|
||||
class CollectionGraphTypeFactory:
|
||||
@classmethod
|
||||
def get(cls, node_type: Type[StrawberryProtocol]) -> Type:
|
||||
if node_type in cls._cache:
|
||||
return cls._cache[node_type]
|
||||
type_name = f"{node_type.__name__.replace('GraphType', '')}Collection"
|
||||
|
||||
if TypeCollector.has(type_name):
|
||||
return TypeCollector.get(type_name)
|
||||
|
||||
node_t = get_provider().get_service(node_type)
|
||||
if not node_t:
|
||||
@@ -21,24 +23,30 @@ class CollectionGraphTypeFactory:
|
||||
|
||||
gql_node = node_t.to_strawberry() if hasattr(node_type, "to_strawberry") else node_type
|
||||
|
||||
gql_type = strawberry.type(
|
||||
type(
|
||||
f"{node_type.__name__.replace("GraphType", "")}Collection",
|
||||
gql_cls = type(
|
||||
type_name,
|
||||
(),
|
||||
{
|
||||
"__annotations__": {
|
||||
{}
|
||||
)
|
||||
|
||||
TypeCollector.set(type_name, gql_cls)
|
||||
|
||||
gql_cls.__annotations__ = {
|
||||
"nodes": List[gql_node],
|
||||
"total_count": int,
|
||||
"count": int,
|
||||
}
|
||||
},
|
||||
)
|
||||
)
|
||||
for k in gql_cls.__annotations__.keys():
|
||||
setattr(gql_cls, k, strawberry.field())
|
||||
|
||||
cls._cache[node_type] = gql_type
|
||||
gql_type = strawberry.type(gql_cls)
|
||||
|
||||
TypeCollector.set(type_name, gql_type)
|
||||
return gql_type
|
||||
|
||||
|
||||
|
||||
|
||||
class Collection:
|
||||
def __init__(self, nodes: list[T], total_count: int, count: int):
|
||||
self._nodes = nodes
|
||||
|
||||
@@ -2,16 +2,16 @@ from typing import Type, Any
|
||||
|
||||
|
||||
class TypeCollector:
|
||||
_registry: dict[type, Type] = {}
|
||||
_registry: dict[type | str, Type] = {}
|
||||
|
||||
@classmethod
|
||||
def has(cls, base: type) -> bool:
|
||||
def has(cls, base: type | str) -> bool:
|
||||
return base in cls._registry
|
||||
|
||||
@classmethod
|
||||
def get(cls, base: type) -> Type:
|
||||
def get(cls, base: type | str) -> Type:
|
||||
return cls._registry[base]
|
||||
|
||||
@classmethod
|
||||
def set(cls, base: type, gql_type: Type):
|
||||
def set(cls, base: type | str, gql_type: Type):
|
||||
cls._registry[base] = gql_type
|
||||
|
||||
Reference in New Issue
Block a user