srht/graphql: Add GraphQLOperation class

Add a new GraphQLOperation class which supports the GraphQL multipart
request spec [1].

[1]: https://github.com/jaydenseric/graphql-multipart-request-spec
This commit is contained in:
Adnan Maolood 2022-06-29 08:07:43 -04:00 committed by Drew DeVault
parent 744c3cf67c
commit d5f6a87972
2 changed files with 79 additions and 24 deletions

View File

@ -1,2 +1,2 @@
from .blueprint import gql_blueprint
from .client import exec_gql, gql_time, DATE_FORMAT, GraphQLError
from .client import exec_gql, gql_time, DATE_FORMAT, GraphQLOperation, GraphQLUpload, GraphQLError

View File

@ -1,3 +1,4 @@
import json
import requests
from datetime import datetime
from flask import request, has_request_context
@ -12,31 +13,85 @@ class GraphQLError(Exception):
self.errors = body["errors"]
def exec_gql(site, query, user=None, client_id=None, valid=None, **variables):
"""
Executes a GraphQL query against the given site's GraphQL API. If no user
is specified, the authenticated user is used. If a validation argument is
provided, the GraphQL response will be interpreted for errors; otherwise
any GraphQL error will cause an exception to be thrown.
"""
origin = cfg(site, "api-origin", default=get_origin(site))
op = GraphQLOperation(query)
for key, value in variables.items():
op.var(key, value)
return op.execute(site, user=user, client_id=client_id, valid=valid)
r = requests.post(f"{origin}/query",
headers={
"X-Forwarded-For": ", ".join(request.access_route) if has_request_context() else None,
**encrypt_request_authorization(user=user, client_id=client_id),
},
json={
"query": query,
"variables": variables,
})
resp = r.json()
if r.status_code != 200 or "errors" in resp:
if valid is None:
raise GraphQLError(r.json())
class GraphQLUpload:
def __init__(self, filename, contents, content_type):
self.filename = filename
self.contents = contents
self.content_type = content_type
class GraphQLOperation:
def __init__(self, query):
self.query = query
self.variables = {}
self.uploads = []
self.map = {}
def var(self, key, value):
assert(key not in self.variables)
if isinstance(value, GraphQLUpload):
self.multipart = True
self.map[str(len(self.uploads))] = [f"variables.{key}"]
self.uploads.append(value)
self.variables[key] = None
elif isinstance(value, list) and all(isinstance(x, GraphQLUpload) for x in value):
self.multipart = True
for i, upload in enumerate(value):
self.map[str(len(self.uploads))] = [f"variables.{key}.{i}"]
self.uploads.append(upload)
self.variables[key] = [None] * len(value)
else:
_copy_errors(valid, resp)
return resp.get("data")
return resp["data"]
self.variables[key] = value
def execute(self, site, user=None, client_id=None, valid=None):
"""
Executes a GraphQL query against the given site's GraphQL API. If no user
is specified, the authenticated user is used. If a validation argument is
provided, the GraphQL response will be interpreted for errors; otherwise
any GraphQL error will cause an exception to be thrown.
"""
origin = cfg(site, "api-origin", default=get_origin(site))
headers={
"X-Forwarded-For": ", ".join(request.access_route) if has_request_context() else None,
**encrypt_request_authorization(user=user, client_id=client_id),
}
if len(self.uploads) > 0:
files = {}
for i, upload in enumerate(self.uploads):
files[str(i)] = (upload.filename, upload.contents, upload.content_type)
r = requests.post(f"{origin}/query",
headers=headers,
files={
'operations': (None, json.dumps({
"query": self.query,
"variables": self.variables,
})),
'map': (None, json.dumps(self.map)),
**files,
})
else:
r = requests.post(f"{origin}/query",
headers=headers,
json={
"query": self.query,
"variables": self.variables,
})
resp = r.json()
if r.status_code != 200 or "errors" in resp:
if valid is None:
raise GraphQLError(r.json())
else:
_copy_errors(valid, resp)
return resp.get("data")
return resp["data"]
def gql_time(time):
"""