Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
LanguageType, required=True, resolver=submission_language_resolver)
source_code = graphene.Field(
graphene.String, required=True, resolver=submission_source_code_resolver)
verdict = graphene.Field(graphene.String, required=True)
@staticmethod
def resolve_verdict(root, _info):
return root.verdict
class Meta:
model = Submission
only_fields = ('id', 'problem', 'issued_time', 'issuer')
class SubmissionQuery(graphene.ObjectType):
submission = graphene.NonNull(SubmissionType, id=graphene.Int(
required=True), resolver=submission_resolver)
submissions = graphene.NonNull(graphene.List(
graphene.NonNull(SubmissionType)), resolver=submissions_resolver)
class SubmitSolution(graphene.Mutation):
class Arguments:
problem_id = graphene.ID(required=True)
language_id = graphene.ID(required=True)
source_code = Upload(required=True)
Output = SubmissionType
mutate = submit_solution_mutate
class SubmissionMutation(graphene.ObjectType):
submit_solution = SubmitSolution.Field()
def convert_postgres_range_to_string(
field, registry=None, input_flag=None, nested_field=False
):
inner_type = convert_django_field(field.base_field)
if not isinstance(inner_type, (List, NonNull)):
inner_type = type(inner_type)
return List(
inner_type,
description=field.help_text or field.verbose_name,
required=is_required(field) and input_flag == "create",
)
#: If any permission should allow the user to execute this mutation
permissions_any = None
#: If we should allow unauthenticated users to do this mutation
allow_unauthenticated = False
class BaseMutation(ClientIDMutation):
"""Base mutation enchanced with permission checking and relay id handling."""
class Meta:
abstract = True
#: A list of errors that happened during the mutation
errors = graphene.List(
graphene.NonNull(MutationErrorType),
description="List of errors that occurred while executing the mutation.",
)
@classmethod
def __init_subclass_with_meta__(cls, permissions=None,
permissions_any=True,
allow_unauthenticated=False,
_meta=None, **kwargs):
if not _meta:
_meta = BaseMutationOptions(cls)
_meta.permissions = permissions or []
_meta.permissions_any = permissions_any
_meta.allow_unauthenticated = allow_unauthenticated
super().__init_subclass_with_meta__(_meta=_meta, **kwargs)
post.status = Post.STATUS_ALMOST
post.title = title or post.title
post.save()
tasks.publish_post_task.delay(post.id, title)
return PublishPost(accepted=True)
class PostInput(graphene.InputObjectType):
id = graphene.ID(required=True)
class RejectPosts(graphene.Mutation):
count = graphene.Int()
class Arguments:
posts = graphene.List(graphene.NonNull(graphene.ID), required=True)
@permissions(is_staff)
def mutate(self, info: ResolveInfo, posts):
post_ids = [
relay.Node.from_global_id(p)[1]
for p in posts
]
posts = Post.objects.filter(pk__in=post_ids)
rows = posts.update(status=Post.STATUS_REJECTED)
return RejectPosts(count=rows)
class ToggleRedditState(graphene.Mutation):
enabled = graphene.Boolean()
class Arguments:
else:
invert_values = lambda arr: 1 - arr
return lambda df: np.where(
df["predictions_agree"],
df[non_principal_prediction_label],
invert_values(df[non_principal_prediction_label]),
)
class RoundPredictionType(graphene.ObjectType):
"""Official Tipresias predictions for a given round."""
round_number = graphene.NonNull(graphene.Int)
match_predictions = graphene.List(
graphene.NonNull(MatchPredictionType), required=True
)
@staticmethod
def resolve_match_predictions(root: RoundPredictions, _info) -> MatchPredictions:
"""Return prediction data for matches in the given round."""
predictions = pd.DataFrame(
root["match_predictions"]
.prefetch_related("match", "ml_model", "match__teammatch_set")
.values(
"match__id",
"match__start_date_time",
"ml_model__is_principal",
"predicted_winner__name",
"predicted_margin",
"predicted_win_probability",
def __init__(
self, field_name: str, field_type: type = None, required=None, **kwargs
):
# Initiate and get specific field info.
self.field_name = field_name
self.field_type = field_type
self.field_source = kwargs.get("source", field_name)
# Add support for NonNull/required fields
if required:
self.field_type = graphene.NonNull(field_type)
# Legacy collection API (Allow lists):
self.extract_key = kwargs.get("key", None)
is_list = kwargs.get("is_list", False)
if is_list:
self.field_type = graphene.List(field_type)
def non_null_list(ttype):
return graphene.NonNull(graphene.List(graphene.NonNull(ttype)))
def convert_field_to_boolean(field, registry=None, input_flag=None, nested_field=False):
required = is_required(field) and input_flag == "create"
if required:
return NonNull(Boolean, description=field.help_text)
return Boolean(description=field.help_text)
from os import devnull
from random import randrange
import falcon
import graphene
# Define a GraphQL query schema
class Query(graphene.ObjectType):
"A Graphene schema. Utilized by GraphQLResource."
hello = graphene.String(description='A basic GraphQL object.')
extra = graphene.String(description='Another basic GraphQL object.')
roll_dice = graphene.List(description='A GraphQL object with arguments.',
of_type=graphene.Int,
args=dict(
dice=graphene.NonNull(of_type=graphene.Int),
sides=graphene.Int())
)
def resolve_hello(self, args, context, info):
return 'Hello world!'
def resolve_extra(self, args, context, info):
return 'Extra!'
def resolve_roll_dice(self, args, context, info):
num_dice = range(args.get('dice'))
return [randrange(1, args.get('sides', 6) + 1) for i in num_dice]
# Create the schema that will be used to resolve GraphQL requests.
schema = graphene.Schema(query=Query)
def convert_ndb_scalar_property(graphene_type, ndb_prop, registry=None, **kwargs):
kwargs['description'] = "%s %s property" % (ndb_prop._name, graphene_type)
_type = graphene_type
if ndb_prop._repeated:
_type = List(_type)
if ndb_prop._required:
_type = NonNull(_type)
return Field(_type, **kwargs)