Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if chatroom.user.id != user.id:
raise ValidationError(
_("You are not the creator of this chatroom"))
if description is not None:
chatroom.description = description
if private is not None:
chatroom.private = private
chatroom.save()
return UpdateChatRoom(chatroom=chatroom)
# {{{2 UpdateHint
class UpdateHint(relay.ClientIDMutation):
hint = graphene.Field(HintNode)
class Input:
content = graphene.String()
hintId = graphene.Int()
@classmethod
def mutate_and_get_payload(cls, root, info, **input):
user = info.context.user
if (not user.is_authenticated):
raise ValidationError(_("Please login!"))
hintId = input['hintId']
content = input['content']
if not content:
"Checks if entered thread name already exists"
if ThreadModel.objects.filter(name=cleaned_input.get('name')).first():
raise GraphQLError('Thread name already taken.')
# Creating thread
new_thread = ThreadModel(**cleaned_input)
new_thread.save()
# Adding creator to the thread
new_membership = ThreadMemberModel(user=called_user, thread=new_thread, is_admin=True)
new_membership.save()
return CreateThread(thread=new_thread)
class UpdateThread(graphene.relay.ClientIDMutation):
"""
Updates a thread with the optional arguments provided.
"""
class Input:
name = graphene.String(required=True, description="Thread name")
new_name = graphene.String(description="Updated thread name")
description = graphene.String(description="Updated thread description")
' Fields '
thread = graphene.Field(ThreadNode)
def mutate_and_get_payload(root, info, **input):
called_user = info.context.user
if called_user is None or called_user.is_anonymous:
raise GraphQLError('Not logged in.')
' Fields '
successful = graphene.Boolean()
def mutate(self, info):
if info.context.user.is_anonymous:
raise GraphQLError('Not logged in.')
try:
logout(info.context)
except:
raise GraphQLError('Failed to log in.')
return Logout(successful=True)
class UpdateProfile(graphene.relay.ClientIDMutation):
"""
Update user's profile according to the provided optional arguments.
"""
class Input:
username = graphene.String(required=True)
first_name = graphene.String()
last_name = graphene.String()
profile_image = Upload()
' Fields '
user = graphene.Field(UserNode)
def mutate_and_get_payload(root, info, **input):
user = UserModel.objects.get(username=input.pop('username'))
filtered_input = clean_input(input)
'password1': input['password1'],
'password2': input['password2'],
}
response = requests.post(
f"{SERVER_URL}/rest-auth/registration/", data=data
)
if response.status_code != 400:
data = json.loads(response.text)
return Registration(data['key'])
elif response.status_code == 400:
return GraphQLError(response.text)
else:
return None
class ConfirmEmail(relay.ClientIDMutation):
class Input:
key = graphene.String(required=True)
detail = graphene.String()
@classmethod
def mutate_and_get_payload(cls, root, info, **input):
data = { 'key': input['key'] }
response = requests.post(
f"{SERVER_URL}/rest-auth/registration/verify-email/",
data=data
)
if response.status_code != 400:
data = json.loads(response.text)
return ConfirmEmail(data['detail'])
elif response.status_code == 400:
ok = graphene.Boolean()
config = graphene.Field(Configuration)
@classmethod
def mutate_and_get_payload(cls, root, info, my_config, client_mutation_id=None):
new_config = tracker.model.configuration.create_configuration(my_config.config_id, my_config.config)
ok = True
return CreateConfiguration(config=new_config, ok=ok)
class WorkflowInput(graphene.InputObjectType):
workflow_name = graphene.String()
workflow_version = graphene.String()
config_id = graphene.ID(required=True)
class CreateWorkflow(relay.ClientIDMutation):
class Input:
my_workflow = WorkflowInput()
ok = graphene.Boolean()
workflow = graphene.Field(Workflow)
@classmethod
def mutate_and_get_payload(cls, root, info, my_workflow, client_mutation_id=None):
new_workflow = tracker.model.workflow.create_workflow(my_workflow.workflow_name, my_workflow.workflow_version, my_workflow.config_id)
ok = True
return CreateWorkflow(workflow=new_workflow, ok=ok)
class AnalysisInput(graphene.InputObjectType):
analysis_name = graphene.String()
start_date = graphene.String()
config_id = graphene.ID(required=True)
if not is_member(info.context.user, thread):
raise GraphQLError('Not a member.')
"Creating the post"
cleaned_input = clean_input(input)
cleaned_input = dict(cleaned_input, **{
'user' : info.context.user,
'thread': thread
})
post = PostModel(**cleaned_input)
post.save()
return CreatePostOnThread(post=post)
class UpdatePostOnThread(graphene.relay.ClientIDMutation):
"""
Fetches and changes the data of the specified post;
returns the udpated post to user
"""
class Input:
post_unique_identifier = graphene.String(required=True, description="Unique ID of the thread")
title = graphene.String(description="New title")
content = graphene.String(description="New content")
' Fields '
post = graphene.Field(PostNode)
def mutate_and_get_payload(root, info, **input):
if info.context.user.is_anonymous:
raise GraphQLError('Not logged in.')
updated_post = PostModel.objects.get(unique_identifier=input.pop('post_unique_identifier'))
def mutate_and_get_payload(cls, root, info, **input):
data = {
'username': input['username'],
'password': input['password']
}
response = requests.post(f"{SERVER_URL}/rest-auth/login/", data=data)
data = json.loads(response.text)
if response.status_code == 200:
return Login(data['key'])
elif response.status_code == 400:
return GraphQLError(response.text)
else:
return None
class Logout(relay.ClientIDMutation):
class Input:
key = graphene.String(required=True)
detail = graphene.String()
@classmethod
def mutate_and_get_payload(cls, root, info, **input):
response = requests.post(
f"{SERVER_URL}/rest-auth/logout/",
headers={'authorization': f"Token {input['key']}"})
if response.status_code == 200:
data = json.loads(response.text)
return Logout(data['detail'])
elif response.status_code == 400:
return GraphQLError(response.text)
else: