sh-edraft.de/cpl
5
0
Fork 0
Code Issues 5 Pull Requests 2 Actions Packages Projects 1 Releases 12 Wiki Activity

Added gql base #181

Browse Source
This commit is contained in:
Sven Heidemann
2025-09-26 21:56:21 +02:00
parent 685c20e3bf
commit e1ab9cf0db
33 changed files with 500 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
src/cpl-graphql/cpl/graphql/__init__.py Normal file
View File

0
src/cpl-graphql/cpl/graphql/_endpoints/__init__.py Normal file
View File

37
src/cpl-graphql/cpl/graphql/_endpoints/graphiql.py Normal file
View File

@@ -0,0 +1,37 @@
from starlette.responses import HTMLResponse
async def graphiql_endpoint(request):
return HTMLResponse("""
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8" />
<title>GraphiQL</title>
<link href="https://unpkg.com/graphiql@2.4.0/graphiql.min.css" rel="stylesheet" />
</head>
<body style="margin:0;overflow:hidden;">
<div id="graphiql" style="height:100vh;"></div>
<!-- React + ReactDOM -->
<script src="https://unpkg.com/react@18.2.0/umd/react.production.min.js"></script>
<script src="https://unpkg.com/react-dom@18.2.0/umd/react-dom.production.min.js"></script>
<!-- GraphiQL -->
<script src="https://unpkg.com/graphiql@2.4.0/graphiql.min.js"></script>
<script>
const graphQLFetcher = graphQLParams =>
fetch('/api/graphql', {
method: 'post',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(graphQLParams),
}).then(response => response.json()).catch(() => response.text());
ReactDOM.render(
React.createElement(GraphiQL, { fetcher: graphQLFetcher }),
document.getElementById('graphiql'),
);
</script>
</body>
</html>
""")

13
src/cpl-graphql/cpl/graphql/_endpoints/graphql.py Normal file
View File

@@ -0,0 +1,13 @@
from starlette.requests import Request
from starlette.responses import Response, JSONResponse
from cpl.graphql.service.service import GraphQLService
async def graphql_endpoint(request: Request, service: GraphQLService) -> Response:
body = await request.json()
query = body.get("query")
variables = body.get("variables")
response_data = await service.execute(query, variables, request)
return JSONResponse(response_data)

27
src/cpl-graphql/cpl/graphql/_endpoints/playground.py Normal file
View File

@@ -0,0 +1,27 @@
from starlette.requests import Request
from starlette.responses import Response, HTMLResponse
async def playground_endpoint(request: Request) -> Response:
return HTMLResponse("""
<!DOCTYPE html>
<html>
<head>
<meta charset=utf-8/>
<title>GraphQL Playground</title>
<link rel="stylesheet" href="https://unpkg.com/graphql-playground-react/build/static/css/index.css" />
<link rel="shortcut icon" href="https://raw.githubusercontent.com/graphql/graphql-playground/master/packages/graphql-playground-react/public/favicon.png" />
<script src="https://unpkg.com/graphql-playground-react/build/static/js/middleware.js"></script>
</head>
<body>
<div id="root"/>
<script>
window.addEventListener('load', function () {
GraphQLPlayground.init(document.getElementById('root'), {
endpoint: '/api/graphql'
})
})
</script>
</body>
</html>
""")

0
src/cpl-graphql/cpl/graphql/abc/__init__.py Normal file
View File

54
src/cpl-graphql/cpl/graphql/abc/query_base.py Normal file
View File

@@ -0,0 +1,54 @@
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
src/cpl-graphql/cpl/graphql/application/__init__.py Normal file
View File

@@ -0,0 +1 @@
from .graphql_app import WebApp

80
src/cpl-graphql/cpl/graphql/application/graphql_app.py Normal file
View File

@@ -0,0 +1,80 @@
from enum import Enum
from typing import Self
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 .._endpoints.graphiql import graphiql_endpoint
from .._endpoints.graphql import graphql_endpoint
from .._endpoints.playground import playground_endpoint
from ..graphql_module import GraphQLModule
from ..service.schema import Schema
class GraphQLApp(WebApp):
def __init__(self, services: ServiceProvider, modules: Modules):
WebApp.__init__(self, services, modules, [GraphQLModule])
def with_graphql(
self,
authentication: bool = False,
roles: list[str | Enum] = None,
permissions: list[str | Enum] = None,
policies: list[str] = None,
match: ValidationMatch = None,
) -> Schema:
self.with_route(
path="/api/graphql",
fn=graphql_endpoint,
method="POST",
authentication=authentication,
roles=roles,
permissions=permissions,
policies=policies,
match=match,
)
schema = self._services.get_service(Schema)
if schema is None:
self._logger.fatal("Could not resolve RootQuery. Make sure GraphQLModule is registered.")
return schema
def with_graphiql(
self,
authentication: bool = False,
roles: list[str | Enum] = None,
permissions: list[str | Enum] = None,
policies: list[str] = None,
match: ValidationMatch = None,
) -> Self:
self.with_route(
path="/api/graphiql",
fn=graphiql_endpoint,
method="GET",
authentication=authentication,
roles=roles,
permissions=permissions,
policies=policies,
match=match,
)
return self
def with_playground(
self,
authentication: bool = False,
roles: list[str | Enum] = None,
permissions: list[str | Enum] = None,
policies: list[str] = None,
match: ValidationMatch = None,
) -> Self:
self.with_route(
path="/api/playground",
fn=playground_endpoint,
method="GET",
authentication=authentication,
roles=roles,
permissions=permissions,
policies=policies,
match=match,
)
return self

17
src/cpl-graphql/cpl/graphql/graphql_module.py Normal file
View File

@@ -0,0 +1,17 @@
from cpl.api import ApiModule
from cpl.dependency.module.module import Module
from cpl.dependency.service_provider import ServiceProvider
from cpl.graphql.schema.root_query import RootQuery
from cpl.graphql.service.schema import Schema
from cpl.graphql.service.service import GraphQLService
class GraphQLModule(Module):
dependencies = [ApiModule]
singleton = [Schema, RootQuery]
scoped = [GraphQLService]
@staticmethod
def configure(services: ServiceProvider) -> None:
schema = services.get_service(Schema)
schema.build()

0
src/cpl-graphql/cpl/graphql/schema/__init__.py Normal file
View File

30
src/cpl-graphql/cpl/graphql/schema/field.py Normal file
View File

@@ -0,0 +1,30 @@
from cpl.graphql.abc.query_base import QueryBase
class Field:
def __init__(self, name: str, gql_type: str, resolver: callable, args: dict | None = None, subquery: QueryBase | None = None):
self._name = name
self._gql_type = gql_type
self._resolver = resolver
self._args = args or {}
self._subquery: QueryBase | None = subquery
@property
def name(self) -> str:
return self._name
@property
def type(self) -> str:
return self._gql_type
@property
def resolver(self) -> callable:
return self._resolver
@property
def args(self) -> dict:
return self._args
@property
def subquery(self) -> QueryBase | None:
return self._subquery

6
src/cpl-graphql/cpl/graphql/schema/query.py Normal file
View File

@@ -0,0 +1,6 @@
from cpl.graphql.abc.query_base import QueryBase
class Query(QueryBase):
def __init__(self):
QueryBase.__init__(self)

6
src/cpl-graphql/cpl/graphql/schema/root_query.py Normal file
View File

@@ -0,0 +1,6 @@
from cpl.graphql.schema.query import Query
class RootQuery(Query):
def __init__(self):
Query.__init__(self)

0
src/cpl-graphql/cpl/graphql/service/__init__.py Normal file
View File

56
src/cpl-graphql/cpl/graphql/service/schema.py Normal file
View File

@@ -0,0 +1,56 @@
import graphene
from cpl.api import APILogger
from cpl.dependency.service_provider import ServiceProvider
from cpl.graphql.schema.query import Query
from cpl.graphql.schema.root_query import RootQuery
class Schema:
def __init__(self, logger: APILogger, query: RootQuery, provider: ServiceProvider):
self._logger = logger
self._provider = provider
self._query = query
self._schema = None
@property
def schema(self) -> graphene.Schema | None:
return self._schema
@property
def query(self) -> RootQuery:
return self._query
def build(self) -> graphene.Schema:
self._schema = graphene.Schema(
query=self.to_graphene(self._query),
mutation=None,
subscription=None,
)
return self._schema
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:
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
)
class_name = name or query.__class__.__name__
return type(class_name, (graphene.ObjectType,), attrs)

31
src/cpl-graphql/cpl/graphql/service/service.py Normal file
View File

@@ -0,0 +1,31 @@
from typing import Any, Dict, Optional
from cpl.api.typing import TRequest
from cpl.graphql.service.schema import Schema
class GraphQLService:
def __init__(self, schema: Schema):
self._schema = schema.schema
if self._schema is None:
raise ValueError("Schema has not been built. Call schema.build() before using the service.")
async def execute(
self,
query: str,
variables: Optional[Dict[str, Any]],
request: TRequest,
) -> Dict[str, Any]:
result = await self._schema.execute_async(
query,
variable_values=variables,
context_value={"request": request},
)
response_data: Dict[str, Any] = {}
if result.errors:
response_data["errors"] = [str(e) for e in result.errors]
if result.data:
response_data["data"] = result.data
return response_data

5
src/cpl-graphql/cpl/graphql/typing.py Normal file
View File

@@ -0,0 +1,5 @@
from typing import Type
from cpl.graphql.schema.query import Query
TQuery = Type[Query]

30
src/cpl-graphql/pyproject.toml Normal file
View File

@@ -0,0 +1,30 @@
[build-system]
requires = ["setuptools>=70.1.0", "wheel>=0.43.0"]
build-backend = "setuptools.build_meta"
[project]
name = "cpl-database"
version = "2024.7.0"
description = "CPL database"
readme ="CPL database package"
requires-python = ">=3.12"
license = { text = "MIT" }
authors = [
{ name = "Sven Heidemann", email = "sven.heidemann@sh-edraft.de" }
]
keywords = ["cpl", "database", "backend", "shared", "library"]
dynamic = ["dependencies", "optional-dependencies"]
[project.urls]
Homepage = "https://www.sh-edraft.de"
[tool.setuptools.packages.find]
where = ["."]
include = ["cpl*"]
[tool.setuptools.dynamic]
dependencies = { file = ["requirements.txt"] }
optional-dependencies.dev = { file = ["requirements.dev.txt"] }

1
src/cpl-graphql/requirements.dev.txt Normal file
View File

@@ -0,0 +1 @@
black==25.1.0

2
src/cpl-graphql/requirements.txt Normal file
View File

@@ -0,0 +1,2 @@
cpl-api
graphene==3.4.3