Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from helpers.room_filter.room_filter import location_join_room
from helpers.auth.user_details import get_user_from_db
from helpers.auth.admin_roles import admin_roles
class Devices(SQLAlchemyObjectType):
"""
Returns the device payload with the fields
[id: ID!, name: String!, deviceType: String!,
dateAdded: DateTime!, lastSeen: DateTime!
"""
class Meta:
model = DevicesModel
class CreateDevice(graphene.Mutation):
"""
Returns the device payload after creating
"""
class Arguments:
name = graphene.String(required=True)
room_id = graphene.Int(required=True)
device_type = graphene.String(required=True)
device = graphene.Field(Devices)
@Auth.user_roles('Admin', 'Super Admin')
def mutate(self, info, **kwargs):
room_location = location_join_room().filter(
RoomModel.id == kwargs['room_id'],
RoomModel.state == 'active'
).first()
if not room_location:
is_active=False
)
user.set_password(password)
user.save()
if djoser_settings.get('SEND_ACTIVATION_EMAIL'):
send_activation_email(user, info.context)
return Register(success=bool(user.id))
# TODO: specify exception
except Exception:
errors = ["email", "Email already registered."]
return Register(success=False, errors=errors)
errors = ["password", "Passwords don't match."]
return Register(success=False, errors=errors)
class Activate(graphene.Mutation):
"""
Mutation to activate a user's registration
"""
class Arguments:
token = graphene.String(required=True)
uid = graphene.String(required=True)
success = graphene.Boolean()
errors = graphene.List(graphene.String)
def mutate(self, info, token, uid):
try:
uid = decode_uid(uid)
user = UserModel.objects.get(pk=uid)
if not default_token_generator.check_token(user, token):
class Checkpoint(Mutation):
class Meta:
description = 'Tell the suite to checkpoint its current state.'
resolver = partial(mutator, command='take_checkpoints')
class Arguments:
workflows = List(WorkflowID, required=True)
name = String(
description='The checkpoint name.',
required=True
)
result = GenericScalar()
class ExtTrigger(Mutation):
class Meta:
description = sstrip('''
Report an external event message to a suite server program.
It is expected that a task in the suite has registered the same
message as an external trigger - a special prerequisite to be
satisfied by an external system, via this command, rather than by
triggering off other tasks.
The ID argument should uniquely distinguish one external trigger
event from the next. When a task's external trigger is satisfied by
an incoming message, the message ID is broadcast to all downstream
tasks in the cycle point as `$CYLC_EXT_TRIGGER_ID` so that they can
use it - e.g. to identify a new data file that the external
triggering system is responding to.
).first()
if not room_location:
raise GraphQLError("Room not found")
user = get_user_from_db()
device = DevicesModel(
**kwargs,
date_added=datetime.now(),
last_seen=datetime.now(),
location=user.location
)
device.save()
return CreateDevice(device=device)
class UpdateDevice(graphene.Mutation):
"""
Returns the device payload after updating
"""
class Arguments:
device_id = graphene.Int()
name = graphene.String()
device_type = graphene.String()
location = graphene.String()
room_id = graphene.Int(required=True)
device = graphene.Field(Devices)
@Auth.user_roles('Admin', 'Super Admin')
def mutate(self, info, device_id, **kwargs):
validate_empty_fields(**kwargs)
query_device = Devices.get_query(info)
def resolve_all_invoices(self, info, maxSat=None, minSat=None,
status=None):
invoices = cl.listinvoices()['invoices']
if maxSat:
invoices = [i for i in invoices if i['msatoshi'] < maxSat]
if minSat:
invoices = [i for i in invoices if i['msatoshi'] > minSat]
if status:
invoices = [i for i in invoices if i['status'] == status]
return [Invoice(**invoice) for invoice in invoices]
class CreateInvoice(graphene.Mutation):
class Arguments:
msatoshi = graphene.Int()
label = graphene.String()
description = graphene.String()
ok = graphene.Boolean()
invoice = graphene.Field(lambda: Invoice)
def mutate(self, info, msatoshi, label, description):
cl_invoice = cl.invoice(msatoshi, label, description)
# FIXME: should we look invoice up to get more fields?
# invoices = cl.listinvoices(label)['invoices']
# assert len(invoices) == 1
# invoice = Invoice(**invoices[0])
invoice = Invoice(**cl_invoice)
ok = True
if exceeded:
return UpvoteEntry(feedback=reason)
# Usual upvote
upvoted.add(entry)
if sender.is_karma_eligible:
sender.karma = karma - cost
entry.author.karma = karma + upvote_rate
sender.save()
entry.author.save()
return response
class DownvoteEntry(Action, Mutation):
"""Mutation to downvote an entry, same logic with UpvoteEntry but reversed."""
@staticmethod
@voteaction
def mutate(_root, entry, sender, upvoted, downvoted, in_upvoted, in_downvoted, constants, exceeded, reason):
response = DownvoteEntry(feedback=None)
karma, cost, downvote_rate, upvote_rate = constants
# User removes the downvote
if in_downvoted:
downvoted.remove(entry)
if sender.is_karma_eligible:
sender.karma = karma + cost # refund
entry.author.karma = karma + downvote_rate
sender.save()
@staticmethod
def mutate(root, args, context, info):
section_id = args.get('section_id')
section_id = int(Node.from_global_id(section_id)[1])
section = models.Section.get(section_id)
if section.section_type != SectionTypesEnum.CUSTOM.value:
return DeleteSection(success=False)
require_instance_permission(CrudPermissions.DELETE, section, context)
section.db.delete(section)
section.db.flush()
return DeleteSection(success=True)
class UpdateSection(graphene.Mutation):
__doc__ = docs.UpdateSection.__doc__
class Input:
id = graphene.ID(required=True, description=docs.UpdateSection.id)
title_entries = graphene.List(LangStringEntryInput, description=docs.UpdateSection.title_entries)
url = graphene.String(description=docs.UpdateSection.url)
order = graphene.Float(description=docs.UpdateSection.order)
section = graphene.Field(lambda: Section)
@staticmethod
def mutate(root, args, context, info):
cls = models.Section
section_id = args.get('id')
section_id = int(Node.from_global_id(section_id)[1])
section = cls.get(section_id)
def latlng(self):
return "({},{})".format(self.lat, self.lng)
class Address(graphene.ObjectType):
latlng = graphene.String()
class Query(graphene.ObjectType):
address = graphene.Field(Address, geo=GeoInput(required=True))
def resolve_address(self, info, geo):
return Address(latlng=geo.latlng)
class CreateAddress(graphene.Mutation):
class Arguments:
geo = GeoInput(required=True)
Output = Address
def mutate(self, info, geo):
return Address(latlng=geo.latlng)
class Mutation(graphene.ObjectType):
create_address = CreateAddress.Field()
schema = graphene.Schema(query=Query, mutation=Mutation)
query = '''
'Parent Idea does not belong to this discussion') # noqa: E501
else:
# Our thematic, because it inherits from Idea, needs to be
# associated to the root idea of the discussion.
# We create a hidden root thematic, corresponding to the
# phase, child of the discussion root idea,
# and add our thematic as a child of this root thematic.
parent_idea = get_root_thematic_for_phase(phase)
if parent_idea is None:
parent_idea = create_root_thematic(phase)
saobj = create_idea(parent_idea, phase, args, context)
return CreateThematic(thematic=saobj)
class UpdateThematic(graphene.Mutation):
__doc__ = docs.UpdateThematic.__doc__
class Input:
id = graphene.ID(required=True)
title_entries = graphene.List(LangStringEntryInput, description=docs.Default.langstring_entries)
description_entries = graphene.List(LangStringEntryInput, description=docs.Default.langstring_entries)
announcement = graphene.Argument(IdeaAnnouncementInput, description=docs.Idea.announcement)
questions = graphene.List(QuestionInput, description=docs.UpdateThematic.questions)
image = graphene.String(description=docs.Default.image)
order = graphene.Float(description=docs.Default.float_entry)
message_view_override = graphene.String(description=docs.IdeaInterface.message_view_override)
thematic = graphene.Field(lambda: IdeaUnion)
@staticmethod
@abort_transaction_on_exception
import graphene
from app.model import AccountModel
from app.schema.fields import ResponseMessageField
class RegisterMutation(graphene.Mutation):
class Arguments(object):
id = graphene.String()
username = graphene.String()
password = graphene.String()
description = graphene.String()
result = graphene.Field(ResponseMessageField)
@staticmethod
def mutate(root, info, **kwargs):
AccountModel(**kwargs).save()
return RegisterMutation(ResponseMessageField(is_success=True, message="Successfully registered"))