Added possibility to put auth schema to root graphs #181

This commit is contained in:
2025-09-29 08:46:51 +02:00
parent e7e3712e08
commit 5b3872a1fe
5 changed files with 54 additions and 23 deletions

View File

@@ -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,10 +89,23 @@ 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}")
if self._with_graphiql:
self._logger.warning(f"GraphiQL available at http://{self._api_settings.host}:{self._api_settings.port}/api/graphiql")
if self._with_playground:
self._logger.warning(f"GraphQL Playground available at http://{self._api_settings.host}:{self._api_settings.port}/api/playground")
self._logger.warning(f"GraphQL Playground available at http://{self._api_settings.host}:{self._api_settings.port}/api/playground")

View File

@@ -20,4 +20,6 @@ class GraphQLAuthModule(Module):
@staticmethod
def configure(provider: ServiceProvider):
schema = provider.get_service(Schema)
schema.with_type(UserGraphType)
schema.with_type(UserGraphType)

View File

@@ -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",
(),
{
"__annotations__": {
"nodes": List[gql_node],
"total_count": int,
"count": int,
}
},
)
gql_cls = type(
type_name,
(),
{}
)
cls._cache[node_type] = gql_type
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())
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

View File

@@ -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