PK ! l{ { strawberry/__init__.py__version__ = "0.1.0"
from .field import field
from .schema import Schema
from .type import type
from .scalars import ID
PK ! f- - strawberry/constants.pyIS_STRAWBERRY_FIELD = "_is_strawberry_field"
PK !
strawberry/exceptions.py# TODO: add links to docs
from typing import List, Set
class MissingReturnAnnotationError(Exception):
"""The field is missing the return annotation"""
def __init__(self, field_name: str):
message = (
f'Return annotation missing for field "{field_name}", '
"did you forget to add it?"
)
super().__init__(message)
class MissingArgumentsAnnotationsError(Exception):
"""The field is missing the annotation for one or more arguments"""
def __init__(self, field_name: str, arguments: Set[str]):
arguments_list: List[str] = sorted(list(arguments))
if len(arguments_list) == 1:
argument = f'argument "{arguments_list[0]}"'
else:
head = ", ".join(arguments_list[:-1])
argument = f'arguments "{head}" and "{arguments_list[-1]}"'
message = (
f"Missing annotation for {argument} "
f'in field "{field_name}", did you forget to add it?'
)
super().__init__(message)
PK ! 7 7 strawberry/field.pyfrom typing import get_type_hints
from graphql import GraphQLField
from .constants import IS_STRAWBERRY_FIELD
from .exceptions import MissingArgumentsAnnotationsError, MissingReturnAnnotationError
from .type_converter import get_graphql_type_for_annotation
from .utils.inspect import get_func_args
def field(wrap):
setattr(wrap, IS_STRAWBERRY_FIELD, True)
annotations = get_type_hints(wrap)
name = wrap.__name__
if "return" not in annotations:
raise MissingReturnAnnotationError(name)
field_type = get_graphql_type_for_annotation(annotations["return"], name)
function_arguments = set(get_func_args(wrap)) - {"self", "info"}
arguments_annotations = {
key: value
for key, value in annotations.items()
if key not in ["info", "return"]
}
annotated_function_arguments = set(arguments_annotations.keys())
arguments_missing_annotations = function_arguments - annotated_function_arguments
if len(arguments_missing_annotations) > 0:
raise MissingArgumentsAnnotationsError(name, arguments_missing_annotations)
arguments = {
name: get_graphql_type_for_annotation(annotation, name)
for name, annotation in arguments_annotations.items()
}
wrap.field = GraphQLField(field_type, args=arguments, resolve=wrap)
return wrap
PK ! o/ / strawberry/scalars.pyimport typing
ID = typing.NewType("ID", str)
PK ! wV strawberry/schema.pyfrom graphql import GraphQLSchema
# TODO: typings
class Schema(GraphQLSchema):
def __init__(self, query):
super().__init__(query=self._build_query(query))
def _build_query(self, query):
return query.field
PK ! mn n strawberry/server/__init__.pyimport click
from starlette.applications import Starlette
from starlette.templating import Jinja2Templates
import importlib
import uvicorn
import pathlib
import hupper
from .graphql_app import GraphQLApp
app = Starlette()
here = pathlib.Path(__file__).parent
templates_path = here / "templates"
templates = Jinja2Templates(directory=str(templates_path))
app = Starlette(debug=True)
@app.route("/")
async def homepage(request):
return templates.TemplateResponse("playground.html", {"request": request})
@click.command()
@click.argument("module")
def main(module):
reloaded = hupper.start_reloader("strawberry.server.main")
schema_module = importlib.import_module(module)
reloaded.watch_files([schema_module.__file__])
app.add_route("/graphql", GraphQLApp(schema_module.schema))
uvicorn.run(app, host="0.0.0.0", port=8000, log_level="error")
PK ! `
$! ! strawberry/server/graphql_app.pyimport datetime
import functools
import json
import typing
from graphql import graphql
from graphql.error import GraphQLError, format_error as format_graphql_error
from pygments import highlight, lexers
from pygments.formatters import Terminal256Formatter
from starlette import status
from starlette.background import BackgroundTasks
from starlette.concurrency import run_in_threadpool
from starlette.requests import Request
from starlette.responses import HTMLResponse, JSONResponse, PlainTextResponse, Response
from starlette.types import ASGIInstance, Receive, Scope, Send
from .utils.graphql_lexer import GraphqlLexer
class GraphQLApp:
def __init__(self, schema) -> None:
self.schema = schema
def __call__(self, scope: Scope) -> ASGIInstance:
return functools.partial(self.asgi, scope=scope)
async def asgi(self, receive: Receive, send: Send, scope: Scope) -> None:
request = Request(scope, receive=receive)
response = await self.handle_graphql(request)
await response(receive, send)
def _debug_log(
self, operation_name: str, query: str, variables: typing.Dict["str", typing.Any]
):
if operation_name == "IntrospectionQuery":
return
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"[{now}]: {operation_name or 'No operation name'}")
print(highlight(query, GraphqlLexer(), Terminal256Formatter()))
if variables:
variables_json = json.dumps(variables, indent=4)
print(highlight(variables_json, lexers.JsonLexer(), Terminal256Formatter()))
async def handle_graphql(self, request: Request) -> Response:
if request.method == "POST":
content_type = request.headers.get("Content-Type", "")
if "application/json" in content_type:
data = await request.json()
elif "application/graphql" in content_type:
body = await request.body()
text = body.decode()
data = {"query": text}
elif "query" in request.query_params:
data = request.query_params
else:
return PlainTextResponse(
"Unsupported Media Type",
status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
)
else:
return PlainTextResponse(
"Method Not Allowed", status_code=status.HTTP_405_METHOD_NOT_ALLOWED
)
try:
query = data["query"]
variables = data.get("variables")
operation_name = data.get("operationName")
except KeyError:
return PlainTextResponse(
"No GraphQL query found in the request",
status_code=status.HTTP_400_BAD_REQUEST,
)
self._debug_log(operation_name, query, variables)
background = BackgroundTasks()
context = {"request": request, "background": background}
result = await self.execute(
query, variables=variables, context=context, operation_name=operation_name
)
error_data = (
[format_graphql_error(err) for err in result.errors]
if result.errors
else None
)
response_data = {"data": result.data, "errors": error_data}
status_code = (
status.HTTP_400_BAD_REQUEST if result.errors else status.HTTP_200_OK
)
return JSONResponse(
response_data, status_code=status_code, background=background
)
async def execute(self, query, variables=None, context=None, operation_name=None):
return await graphql(
self.schema,
query,
variable_values=variables,
operation_name=operation_name,
context_value=context,
)
PK ! ѩK K + strawberry/server/templates/playground.html
GraphQL Playground
Loading
GraphQL Playground
PK ! # strawberry/server/utils/__init__.pyPK ! i ( strawberry/server/utils/graphql_lexer.pyfrom pygments import token
from pygments.lexer import RegexLexer
class GraphqlLexer(RegexLexer):
name = "GraphQL"
aliases = ["graphql", "gql"]
filenames = ["*.graphql", "*.gql"]
mimetypes = ["application/graphql"]
tokens = {
"root": [
(r"#.*", token.Comment.Singline),
(r"\.\.\.", token.Operator),
(r'"[\u0009\u000A\u000D\u0020-\uFFFF]*"', token.String.Double),
(
r"(-?0|-?[1-9][0-9]*)(\.[0-9]+[eE][+-]?[0-9]+|\.[0-9]+|[eE][+-]?[0-9]+)",
token.Number.Float,
),
(r"(-?0|-?[1-9][0-9]*)", token.Number.Integer),
(r"\$+[_A-Za-z][_0-9A-Za-z]*", token.Name.Variable),
(r"[_A-Za-z][_0-9A-Za-z]+\s?:", token.Text),
(r"(type|query|mutation|@[a-z]+|on|true|false|null)\b", token.Keyword.Type),
(r"[!$():=@\[\]{|}]+?", token.Punctuation),
(r"[_A-Za-z][_0-9A-Za-z]*", token.Keyword),
(r"(\s|,)", token.Text),
]
}
PK ! E_ strawberry/type.pyimport typing
from dataclasses import dataclass
from graphql import GraphQLField, GraphQLObjectType
from graphql.utilities.schema_printer import print_type
from .constants import IS_STRAWBERRY_FIELD
from .type_converter import get_graphql_type_for_annotation
def _get_resolver(cls, field_name):
def _resolver(obj, info):
# TODO: can we make this nicer?
# does it work in all the cases?
field_resolver = getattr(cls(**(obj.__dict__ if obj else {})), field_name)
if getattr(field_resolver, IS_STRAWBERRY_FIELD, False):
return field_resolver(obj, info)
return field_resolver
return _resolver
def _get_fields(cls):
cls_annotations = typing.get_type_hints(cls)
fields = {
key: GraphQLField(
get_graphql_type_for_annotation(value, field_name=key),
resolve=_get_resolver(cls, key),
)
for key, value in cls_annotations.items()
}
fields.update(
{
key: value.field
for key, value in cls.__dict__.items()
if getattr(value, IS_STRAWBERRY_FIELD, False)
}
)
return fields
def type(cls):
def wrap():
def repr_(self):
return print_type(self.field)
setattr(cls, "__repr__", repr_)
cls._fields = _get_fields(cls)
cls.field = GraphQLObjectType(name=cls.__name__, fields=cls._fields)
return dataclass(cls, repr=False)
return wrap()
PK ! F脪 strawberry/type_converter.pyfrom graphql import (
GraphQLBoolean,
GraphQLFloat,
GraphQLID,
GraphQLInt,
GraphQLList,
GraphQLNonNull,
GraphQLString,
GraphQLUnionType,
)
from .scalars import ID
TYPE_MAP = {
str: GraphQLString,
int: GraphQLInt,
float: GraphQLFloat,
bool: GraphQLBoolean,
ID: GraphQLID,
}
# TODO: make so that we don't pass force optional
# we use that when trying to get the type for a
# option field (which can either be a scalar or an object type)
def get_graphql_type_for_annotation(
annotation, field_name: str, force_optional: bool = False
):
# TODO: nice error
is_optional = False
# TODO: this might lead to issues with types that have a field value
if hasattr(annotation, "field"):
graphql_type = annotation.field
else:
annotation_name = getattr(annotation, "_name", None)
if annotation_name == "List":
list_of_type = get_graphql_type_for_annotation(
annotation.__args__[0], field_name
)
return GraphQLList(list_of_type)
# for some reason _name is None for Optional and Union types, so we check if we
# have __args__ populated, there might be some edge cases where __args__ is
# populated but the type is not an Union, like in the above case with Lists
if hasattr(annotation, "__args__"):
types = annotation.__args__
non_none_types = [x for x in types if x != type(None)] # noqa:E721
# optionals are represented as Union[type, None]
if len(non_none_types) == 1:
is_optional = True
graphql_type = get_graphql_type_for_annotation(
non_none_types[0], field_name, force_optional=True
)
else:
is_optional = type(None) in types
# TODO: union types don't work with scalar types
# so we want to return a nice error
# also we want to make sure we have been passed
# strawberry types
graphql_type = GraphQLUnionType(
field_name, [type.field for type in types]
)
else:
graphql_type = TYPE_MAP.get(annotation)
if not graphql_type:
raise ValueError(f"Unable to get GraphQL type for {annotation}")
if is_optional or force_optional:
return graphql_type
return GraphQLNonNull(graphql_type)
PK ! strawberry/utils/__init__.pyPK ! "3wQW W strawberry/utils/inspect.pyimport inspect
from typing import Any, Callable
def get_func_args(func: Callable[[Any], Any]):
"""Returns a list of arguments for the function"""
sig = inspect.signature(func)
return [
arg_name
for arg_name, param in sig.parameters.items()
if param.kind == inspect.Parameter.POSITIONAL_OR_KEYWORD
]
PK !Hy;/ 1 3 strawberry_graphql-0.1.0.dist-info/entry_points.txtN+I/N.,()*N-*K--.)J,OJ-*ԃX&fqq PK ! ͪN0 0 * strawberry_graphql-0.1.0.dist-info/LICENSEMIT License
Copyright (c) 2018 Patrick Arminio
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
PK !HnH T U ( strawberry_graphql-0.1.0.dist-info/WHEEL
A
н#Z;/"d&F[xzw@ Zpy3Fv]\fi4WZ^EgM_-]#0(q7 PK !H? @ + strawberry_graphql-0.1.0.dist-info/METADATAMO0 9F
$Ja|"9)OB 8c
4*FC.4 ;Fr҈zl4
;jr$/g6bk_)nh.*P;(oV##%w(J?EÈ飔:ietV ظ|8dt4x
Vl1rԯ/++D7 eiU)w
V_~_
0t2`|F
qb.imk ` FLWKPhH8; YDZ Pi YPK !H]r ) strawberry_graphql-0.1.0.dist-info/RECORDKH-ͥX7n`qQU"ɌzғbV<[ GE]0h'-~74%D]xf2d3ZgXA3/45Qь3MNSfܣIfΝ^
9 e8)S6/kG$i
(q\@(k Ō)efxKcCn9
̤P{KՍ&i@e d6A=^9WMv Z@~VhF: kNry[BO&k*H>+%Yr< w'~qew#8zUkPs{ ߠY yWQ|LSP#|:aqO,q iOVma>_){}qU3O)1ҲP'_BK 9w˵x$-2wqQ9UX^DQ$V݁l]gͽrOP )N)# !Y_P=\w]$ka/xjI#R{&<4wKsT #Ƥ9酺:
xT>_/Y2ՕP,@:InW\t 4WXԨ ʭE6}
MPa'LP>Eވ7K I} ǥ@,5~=m ;jEvx';N]^a W- @IP4yJ=:-DۅNu.@xm$"Xi AúI+k69A'cd*evH oiJNSs]Pĵ3BQ'G,}h%/~jU PK ! l{ { strawberry/__init__.pyPK ! f- - strawberry/constants.pyPK !
strawberry/exceptions.pyPK ! 7 7 Q strawberry/field.pyPK ! o/ /
strawberry/scalars.pyPK ! wV strawberry/schema.pyPK ! mn n 7 strawberry/server/__init__.pyPK ! `
$! ! strawberry/server/graphql_app.pyPK ! ѩK K + ? strawberry/server/templates/playground.htmlPK ! # j strawberry/server/utils/__init__.pyPK ! i ( j strawberry/server/utils/graphql_lexer.pyPK ! E_ o strawberry/type.pyPK ! F脪 u strawberry/type_converter.pyPK ! ~ strawberry/utils/__init__.pyPK ! "3wQW W * strawberry/utils/inspect.pyPK !Hy;/ 1 3 strawberry_graphql-0.1.0.dist-info/entry_points.txtPK ! ͪN0 0 * : strawberry_graphql-0.1.0.dist-info/LICENSEPK !HnH T U ( strawberry_graphql-0.1.0.dist-info/WHEELPK !H? @ + L strawberry_graphql-0.1.0.dist-info/METADATAPK !H]r ) ԇ strawberry_graphql-0.1.0.dist-info/RECORDPK