Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@classmethod
def mutate(cls, instance, args, info):
character = NdbCharacter.create(**args)
return cls(
character=Character.from_ndb_entity(character),
ok=bool(character)
)
class CreateFriendship(graphene.Mutation):
class Input:
character_a = graphene.NonNull(NdbKey)
character_b = graphene.NonNull(NdbKey)
ok = graphene.Boolean().NonNull
@classmethod
def mutate(cls, instance, args, info):
friendship = NdbFriendship.create_from_keys(args['character_a'], args['character_b'])
return cls(ok=bool(friendship))
class Mutation(graphene.ObjectType):
create_character = graphene.Field(CreateCharacter)
create_friendship = graphene.Field(CreateFriendship)
workflows = List(String, required=True)
command = String(
description="""String options:
- `set_stop_cleanly` (default)
- `set_stop_after_clock_time`
- `set_stop_after_point`
- `set_stop_after_task`
- `stop_now`""",
default_value='set_stop_cleanly',)
args = StopWorkflowArgs()
result = GenericScalar()
class TaskArgs(InputObjectType):
check_syntax = Boolean(description="""Use with actions:
- `dry_run_tasks`""")
no_check = Boolean(description="""Use with actions:
- `insert_tasks`""")
stop_point_string = String(description="""Use with actions:
- `insert_tasks`""")
poll_succ = Boolean(description="""Use with actions:
- `poll_tasks`""")
spawn = Boolean(description="""Use with actions:
- `remove_tasks`""")
state = String(description="""Use with actions:
- `reset_task_states`""")
outputs = List(String, description="""Use with actions:
- `reset_task_states`""")
back_out = Boolean(description="""Use with actions:
- `trigger_tasks`""")
)
self._dagster_type = dagster_type
def resolve_type_attributes(self, _info):
return self._dagster_type.type_attributes
def resolve_fields(self, _info):
return [TypeField(name=k, field=v) for k, v in self._dagster_type.field_dict.items()]
class TypeField(graphene.ObjectType):
name = graphene.NonNull(graphene.String)
description = graphene.String()
type = graphene.NonNull(lambda: Type)
default_value = graphene.String()
is_optional = graphene.NonNull(graphene.Boolean)
def __init__(self, name, field):
super(TypeField, self).__init__(
name=name,
description=field.description,
default_value=field.default_value_as_str if field.default_provided else None,
is_optional=field.is_optional
)
self._field = field
def resolve_type(self, _info):
return Type.from_dagster_type(dagster_type=self._field.dagster_type)
class ExecutionPlan(graphene.ObjectType):
steps = non_null_list(lambda: ExecutionStep)
return Attitude.neutral
vote = get_object_or_None(AbstractReplyVote, reply=self, record_user=user)
if vote is None:
return Attitude.neutral
return vote.attitude
def resolve_discussion(self, info, *args, **kwargs):
return list(AbstractReply.objects.filter(ancestor=self.pk))
def resolve_disable(self, info, *args, **kwargs):
return self.disable
class UpdateAbstractReplyVote(graphene.Mutation):
class Arguments:
attitude = graphene.Boolean(required=True)
reply_pk = graphene.ID(required=True)
result = graphene.String()
@login_required
def mutate(self, info, reply_pk, attitude):
reply = AbstractReply.objects.get(pk=reply_pk)
node, created = AbstractReplyVote.objects.get_or_create(
user=info.context.user,
discussion=i
)
attitude = Attitude.agree if attitude else Attitude.disagree
node.vote = attitude if created or attitude != node.vote else Attitude.neutral
node.save()
return UpdateReplyVote(result=attitude)
def convert_form_field_to_boolean(field):
return Boolean(description=field.help_text, required=True)
ids = graphene.List(
graphene.ID, required=True, description="List of page IDs to delete."
)
class Meta:
description = "Deletes pages."
model = models.Page
permissions = (PagePermissions.MANAGE_PAGES,)
class PageBulkPublish(BaseBulkMutation):
class Arguments:
ids = graphene.List(
graphene.ID, required=True, description="List of page IDs to (un)publish."
)
is_published = graphene.Boolean(
required=True, description="Determine if pages will be published or not."
)
class Meta:
description = "Publish pages."
model = models.Page
permissions = (PagePermissions.MANAGE_PAGES,)
@classmethod
def bulk_action(cls, queryset, is_published):
queryset.update(is_published=is_published)
# if we're in ordering mode, we want
# to return True for all CalumaOrdering types,
# and if it's false, we want the opposite
return ordering == isinstance(filt, CalumaOrdering)
filter_fields = {
name: _get_or_make_field(name, filt)
for name, filt in _filter_coll.filters.items()
# exclude orderBy in our fields. We want only new-style order filters
if _should_include_filter(filt) and name != "orderBy"
}
if ordering:
filter_fields["direction"] = AscDesc(default=AscDesc.ASC, required=False)
else:
filter_fields["invert"] = graphene.Boolean(required=False, default=False)
filter_type = type(field_type_name, (InputObjectType,), filter_fields)
converted = List(filter_type)
registry.register_converted_field(field, converted)
return converted
class Question(Node, graphene.Interface):
id = graphene.ID(required=True)
created_at = graphene.DateTime(required=True)
modified_at = graphene.DateTime(required=True)
created_by_user = graphene.String()
created_by_group = graphene.String()
slug = graphene.String(required=True)
label = graphene.String(required=True)
info_text = graphene.String()
is_required = QuestionJexl(
required=True,
description="Required expression is only evaluated when question is not hidden.",
)
is_hidden = QuestionJexl(required=True)
is_archived = graphene.Boolean(required=True)
meta = generic.GenericScalar(required=True)
forms = DjangoFilterConnectionField(
"caluma.caluma_form.schema.Form", filterset_class=filters.FormFilterSet
)
source = graphene.Field("caluma.caluma_form.schema.Question")
@classmethod
def get_queryset(cls, queryset, info):
queryset = super().get_queryset(queryset, info)
return (
queryset.select_related("sub_form", "row_form")
.order_by("-formquestion__sort")
.distinct()
)
@classmethod