Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from django.contrib.gis import geos
import graphene
import graphql_geojson
from . import models, types
from .testcases import SchemaTestCase
class CreatePlace(graphene.Mutation):
place = graphene.Field(types.PlaceType)
class Arguments:
name = graphene.String(required=True)
location = graphql_geojson.Geometry(required=True)
@classmethod
def mutate(cls, root, info, **args):
place = models.Place.objects.create(**args)
return cls(place=place)
class MutationsTests(SchemaTestCase):
query = '''
mutation CreatePlace($name: String!, $location: Geometry!) {
createPlace(name: $name, location: $location) {
place {
id
type
geometry {
import graphene
from graphql import GraphQLError
from backend.likes.models import Like as LikeModel
from backend.posts.models import Post as PostModel
from backend.comments.models import Comment as CommentModel
from backend.likes.schemas.queries import LikeNode
class LikePost(graphene.Mutation):
"""
Adds a like to the specified post.
"""
class Arguments:
post_unique_identifier = graphene.String(required=True, description="Unique identifier of the post")
' Fields '
like = graphene.Field(LikeNode)
def mutate(self, info, post_unique_identifier):
if info.context.user.is_anonymous:
raise GraphQLError('Not logged in.')
post = PostModel.objects.get(unique_identifier=post_unique_identifier)
try:
new_like = LikeModel(content_object=post, user=info.context.user)
except Exception as e:
raise GraphQLError(e)
else:
new_like.save()
return LikePost(like=new_like)
start_date = args.get('start_date', None)
end_date = args.get('end_date', None)
if start_date:
discussion.active_start_date = start_date.replace(tzinfo=None)
if end_date:
discussion.active_end_date = end_date.replace(tzinfo=None)
db.flush()
discussion = cls.get(discussion_id)
return UpdateDiscussion(discussion=discussion)
class LocalePreference(graphene.ObjectType):
__doc__ = docs.LocalePreference.__doc__
locale = graphene.String(description=docs.LocalePreference.locale)
name = graphene.String(in_locale=graphene.String(required=True), description=docs.LocalePreference.name)
native_name = graphene.String(description=docs.LocalePreference.native_name)
def resolve_name(self, args, context, info):
in_locale = args.get('in_locale') or None
locale_model = models.Locale.get_or_create(in_locale)
name = models.LocaleLabel.names_of_locales_in_locale([self.locale],
locale_model)
if not name:
# If the locale label does not exist, fallback on English
locale_model = models.Locale.get_or_create('en')
name = models.LocaleLabel.names_of_locales_in_locale([self.locale],
locale_model)
return name[self.locale]
def resolve_title(stop_info, info):
return stop_info.title
def resolve_lat(stop_info, info):
return stop_info.lat
def resolve_lon(stop_info, info):
return stop_info.lon
class RouteConfig(ObjectType):
id = String()
title = String()
directions = List(DirectionInfo)
stops = List(StopInfo)
stopInfo = Field(StopInfo, stopId = String())
directionInfo = Field(DirectionInfo, directionId = String())
# `parent` is a routeconfig.RouteConfig object
def resolve_id(route_config, info):
return route_config.id
def resolve_title(route_config, info):
return route_config.title
def resolve_stopInfo(route_config, info, stopId):
return route_config.get_stop_info(stopId)
def resolve_directionInfo(route_config, info, directionId):
return route_config.get_direction_info(directionId)
interfaces = (Item, )
user_id = graphene.String()
access_key = graphene.String()
secret_key = graphene.String()
is_active = graphene.Boolean()
is_admin = graphene.Boolean()
resource_policy = graphene.String()
created_at = GQLDateTime()
last_used = GQLDateTime()
concurrency_used = graphene.Int()
rate_limit = graphene.Int()
num_queries = graphene.Int()
user = graphene.UUID()
ssh_public_key = graphene.String()
vfolders = graphene.List('ai.backend.manager.models.VirtualFolder')
compute_sessions = graphene.List(
'ai.backend.manager.models.ComputeSession',
status=graphene.String(),
)
# Deprecated
concurrency_limit = graphene.Int(
deprecation_reason='Moved to KeyPairResourcePolicy object as '
'max_concurrent_sessions field.')
@classmethod
def from_row(
cls,
context: Mapping[str, Any],
comment=item['comment']
)
etc_services = graphene.List(EtcServices)
def resolve_etc_services(self, args, context, info):
for item in query.run('select * from etc_services'):
yield EtcServices(
name=item['name'],
port=item['port'],
protocol=item['protocol'],
aliases=item['aliases'],
comment=item['comment']
)
hash = graphene.List(Hash, directory=graphene.String(),
path=graphene.String()
)
def resolve_hash(self, args, context, info):
if args.get('directory'): where = 'directory = \\"%s\\"' % utils.sanitize(args.get('directory'))
if args.get('path'): where = 'path = \\"%s\\"' % utils.sanitize(args.get('path'))
for item in query.run('select * from hash where %s' % where):
yield Hash(
path=item['path'],
directory=item['directory'],
md5=item['md5'],
sha1=item['sha1'],
sha256=item['sha256']
)
interface_addresses = graphene.List(InterfaceAddresses)
temp = Tenant()
if not_none(group):
temp.group = TenantGroup.objects.get(pk=from_global_id(group)[1])
return NewTenant(tenant=set_and_save(fields, input, temp))
class UpdateTenant(ClientIDMutation):
tenant = Field(TenantNode)
class Input:
id = String(default_value=None)
name = String(default_value=None)
slug = String(default_value=None)
group = String(default_value=None)
description = String(default_value=None)
comments = String(default_value=None)
custom_field_values = String(default_value=None)
@classmethod
def mutate_and_get_payload(cls, input, context, info):
group = input.get('group')
temp = Tenant.objects.get(pk=from_global_id(input.get('id'))[1])
if not_none(group):
temp.group = TenantGroup.objects.get(pk=from_global_id(group)[1])
fields = [ 'name', 'slug', 'description', 'comments', 'custom_field_values' ]
return UpdateTenant(tenant=set_and_save(fields, input, temp))
class DeleteTenant(ClientIDMutation):
tenant = Field(TenantNode)
if skip:
qs = qs[skip::]
if first:
qs = qs[:first]
return qs
def resolve_votes(self, info, **kwargs):
return Vote.objects.all()
class CreateLink(graphene.Mutation):
id = graphene.Int()
url = graphene.String()
description = graphene.String()
posted_by = graphene.Field(UserType)
class Arguments:
url = graphene.String()
description = graphene.String()
def mutate(self, info, url, description):
user = info.context.user
link = Link(
url=url,
description=description,
posted_by=user,
)
link.save()
return await db_conn.scalar(query)
else:
raise TypeError('unexpected type for group_name_or_id')
class Group(graphene.ObjectType):
id = graphene.UUID()
name = graphene.String()
description = graphene.String()
is_active = graphene.Boolean()
created_at = GQLDateTime()
modified_at = GQLDateTime()
domain_name = graphene.String()
total_resource_slots = graphene.JSONString()
allowed_vfolder_hosts = graphene.List(lambda: graphene.String)
integration_id = graphene.String()
scaling_groups = graphene.List(lambda: graphene.String)
@classmethod
def from_row(cls, context: Mapping[str, Any], row: RowProxy) -> Optional[Group]:
if row is None:
return None
return cls(
id=row['id'],
name=row['name'],
description=row['description'],
is_active=row['is_active'],
created_at=row['created_at'],
modified_at=row['modified_at'],
domain_name=row['domain_name'],
total_resource_slots=row['total_resource_slots'].to_json(),
description="""Family cycle instances.""",
args=proxy_args,
resolver=get_nodes_by_ids)
edges = Field(
lambda: Edges,
args=edge_args,
description="""Graph edges""")
nodes_edges = Field(
lambda: NodesEdges,
args=nodes_edges_args,
resolver=get_nodes_edges)
api_version = Int()
cylc_version = String()
last_updated = Float()
meta = Field(DefMeta)
newest_runahead_cycle_point = String()
newest_cycle_point = String()
oldest_cycle_point = String()
reloaded = Boolean()
run_mode = String()
is_held_total = Int()
state_totals = GenericScalar(resolver=resolve_state_totals)
workflow_log_dir = String()
time_zone_info = Field(TimeZone)
tree_depth = Int()
ns_defn_order = List(String)
job_log_names = List(String)
states = List(String)
class Job(ObjectType):
class Meta: