Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
return dt
@staticmethod
def parse_literal(node):
if isinstance(node, graphql.language.ast.StringValue):
return datetime.datetime.strptime(node.value, '%Y-%m-%dT%H:%M:%S.%f')
@staticmethod
def parse_value(value):
return datetime.datetime.strptime(value, '%Y-%m-%dT%H:%M:%S.%f')
# custom compound types
class GeoLocation(graphene.ObjectType):
city = graphene.Field(graphene.String)
continent = graphene.Field(graphene.String)
coordinates = graphene.List(graphene.Float)
country = graphene.Field(graphene.String)
postal_code = graphene.Field(graphene.String)
time_zone = graphene.Field(graphene.String)
@classmethod
def from_ip_address(cls, ip_address):
ip_address = ipaddress.ip_address(ip_address)
if ip_address.is_private:
return
try:
result = geoip.lookup(ip_address)
except geoip2.errors.AddressNotFoundError:
result = None
if result is None:
return
return cls(**result)
def convert_column_to_float(type, column, registry=None):
return Float(description=column.doc, required=not(column.nullable))
from .models import Book, HasRead
class UserType(graphene_django.DjangoObjectType):
is_admin = graphene.Boolean()
def resolve_is_admin(self, info):
return self.is_staff
class Meta(object):
model = UserModel
only_fields = ('id', 'username', 'books_read')
class BookType(graphene_django.DjangoObjectType):
average_rating = graphene.Float()
def resolve_average_rating(self, info):
query = self.read_by.all().aggregate(Avg('rating'))
return query['rating__avg']
class Meta(object):
model = Book
class HasReadType(graphene_django.DjangoObjectType):
class Meta(object):
model = HasRead
class Query(graphene.ObjectType):
users = graphene.List(UserType)
title = String(description="title for the chart")
x_label = String(description="label for the x axis")
y_label = String(description="label for the y axis")
x_low = String(description="lower range for the x axis")
x_high = String(description="higher range for x axis")
y_low = String(description="lower range for y axis")
y_high = String(description="higher range for y axis")
class ParameterDomain(ObjectType):
class Meta:
interfaces = relay.Node,
name = String()
domain_low = Float()
domain_high = Float()
class ParallelCoordinates(ObjectType):
class Meta:
interfaces = relay.Node,
# todo: {key, value}
data = List(ObjectType)
domains = List(ParameterDomain)
class Chart(Union):
"""this is also a file type."""
class Meta:
@classmethod
def mutate_and_get_payload(cls, input, context, info):
temp = Region.objects.get(pk=from_global_id(input.get('id'))[1])
temp.delete()
return DeleteRegion(region=temp)
# Site
class NewSite(ClientIDMutation):
site = Field(SiteNode)
class Input:
name = String()
slug = String()
region = String(default_value=None)
tenant = String(default_value=None)
facility = String(default_value=None)
asn = Float(default_value=None)
physical_address = String(default_value=None)
shipping_address = String(default_value=None)
contact_name = String(default_value=None)
contact_phone = String(default_value=None)
contact_email = String(default_value=None)
comments = String(default_value=None)
@classmethod
def mutate_and_get_payload(cls, input, context, info):
region = input.get('region')
tenant = input.get('tenant')
temp = Site()
if not_none(region):
temp.region = Region.objects.get(pk=from_global_id(region)[1])
import graphene
from .geometry import Geometry
__all__ = ['Distance']
DistanceUnitEnum = graphene.Enum('DistanceUnitEnum', [
(value, value) for value in D.UNITS.keys()
])
class Distance(graphene.InputObjectType):
unit = DistanceUnitEnum(required=True)
value = graphene.Float(required=True)
geometry = Geometry(required=True)
class Meta:
description = """
Distance object type comprising:
\n- checkins: The number of the event checkins in a room
\n- cancellations: The number of the even cacellations in a room
\n- bookings: The number of the room bookings
\n- checkins_percentage: The percentage of room checkins
\n- cancellations_percentage: The percentage of room
cancellations \n- app_bookings: The number of the room bookings
via app \n- app_bookings_percentage: The field with the
percentage of room bookings via app
"""
room_id = graphene.Int()
room_name = graphene.String()
checkins = graphene.Int()
cancellations = graphene.Int()
bookings = graphene.Int()
checkins_percentage = graphene.Float()
cancellations_percentage = graphene.Float()
app_bookings = graphene.Int()
app_bookings_percentage = graphene.Float()
class Calendar(graphene.ObjectType):
"""
Querry for the calendar events and accepts the arguments
\n- events: The calendar events in a room
\n- occupants: The number of occupants of a room
"""
events = graphene.String()
occupants = graphene.String()
class RoomFilter(graphene.ObjectType):
"""
model = MLModel
class CumulativeMetricsByRoundType(graphene.ObjectType):
"""Cumulative performance metrics for the given model through the given round."""
ml_model__name = graphene.String(name="modelName")
cumulative_correct_count = graphene.Int(
description=(
"Cumulative sum of correct tips made by the given model "
"for the given season"
),
default_value=0,
)
cumulative_accuracy = graphene.Float(
description=(
"Cumulative mean of correct tips (i.e. accuracy) made by the given model "
"for the given season."
),
default_value=0,
)
cumulative_mean_absolute_error = graphene.Float(
description="Cumulative mean absolute error for the given season",
default_value=0,
)
cumulative_margin_difference = graphene.Int(
description=(
"Cumulative difference between predicted margin and actual margin "
"for the given season."
),
default_value=0,
permissions = get_permissions(user_id, discussion_id)
vote = vote.handle_duplication(
permissions=permissions, user_id=user_id)
vote.db.add(vote)
proposal.db.flush()
return AddTokenVote(vote_specification=vote_spec)
class AddGaugeVote(graphene.Mutation):
__doc__ = docs.AddGaugeVote.__doc__
class Input:
proposal_id = graphene.ID(required=True)
vote_spec_id = graphene.ID(required=True)
vote_value = graphene.Float(required=False)
vote_specification = graphene.Field(lambda: VoteSpecificationUnion) # need to match GaugeVoteSpecification / NumberGaugeVoteSpecification
@staticmethod
@abort_transaction_on_exception
def mutate(root, args, context, info):
require_cls_permission(CrudPermissions.CREATE, models.TokenIdeaVote, context)
discussion_id = context.matchdict['discussion_id']
discussion = models.Discussion.get(discussion_id)
user_id = context.authenticated_userid
vote_value = args.get('vote_value')
proposal_id = args.get('proposal_id')
proposal_id = int(Node.from_global_id(proposal_id)[1])
proposal = models.Idea.get(proposal_id)
import asyncio
import random
import graphene as G
class RandomType(G.ObjectType):
seconds = G.Int()
random_int = G.Int()
class Subscription(G.ObjectType):
count_seconds = G.Float(up_to=G.Int())
random_int = G.Field(RandomType)
async def resolve_count_seconds(root, info, up_to=5):
for i in range(up_to):
yield i
await asyncio.sleep(1.0)
yield up_to
async def resolve_random_int(root, info):
i = 0
while True:
yield RandomType(seconds=i, random_int=random.randint(0, 500))
await asyncio.sleep(1.0)
i += 1