Python graphene 模块,String() 实例源码
我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用graphene.String()。
def gen_operators_of_field(f_name, mongo_field, r_graphene, operators_list):
""" Return a dict with keys as the name of the field with operator and value is the required type,for instance:
@param f_name: string name of the field
@param mongo_field: object instance of mongoengine field,e.g: mongoengine.StringField()
@param r_graphene: object instance of graphene field,e.g: graphene.String():
{
name: graphene.String()
name__nin: graphene.List(graphene.String) ...
}
"""
field_with_operators = {
f_name: field_to_id(mongo_field, r_graphene)
}
for op_name in operators_list:
field_with_operators[f_name + '__' + op_name] = operators[op_name](mongo_field, r_graphene)
if isinstance(mongo_field, fields_string_operators):
for op in string_operators:
field_with_operators[f_name + '__' + op] = graphene.String()
return field_with_operators
def get_field_args(field):
types = {
str: graphene.String,
bool: graphene.Boolean,
int: graphene.Int,
float: graphene.Float,
}
try:
args = inspect.signature(field)
result = {}
for name, arg in args.parameters.items():
if arg.default is not None:
custom_type = types[type(arg.default)]
else:
custom_type = graphene.String
result.update({name: custom_type()})
return result
except (KeyError, ValueError):
return {}
def schema():
class Query(graphene.ObjectType):
test_string = graphene.String()
def resolve_test_string(self, args, context, info):
return 'works'
# Todo: Implement case conversion for arg names
class Subscription(graphene.ObjectType):
test_subscription = graphene.String()
test_context = graphene.String()
test_filter = graphene.String(filterBoolean=graphene.Boolean())
test_filter_multi = graphene.String(
filterBoolean=graphene.Boolean(),
a=graphene.String(),
b=graphene.Int())
test_channel_options = graphene.String()
def resolve_test_subscription(self, info):
return self
def resolve_test_context(self, info):
return context
def resolve_test_filter(self, info):
return 'good_filter' if args.get('filterBoolean') else 'bad_filter'
def resolve_test_filter_multi(self, info):
return 'good_filter' if args.get('filterBoolean') else 'bad_filter'
def resolve_test_channel_options(self, info):
return self
return graphene.Schema(query=Query, subscription=Subscription)
def test_can_subscribe_to_more_than_one_trigger(sub_mgr):
non_local = {'trigger_count': 0}
query = 'subscription multiTrigger($filterBoolean: Boolean,\
$uga: String){testFilterMulti(filterBoolean: $filterBoolean,\
a: $uga,b: 66)}'
def callback(err, payload):
if err:
sys.exit(err)
else:
try:
if payload is None:
assert True
else:
assert payload.data.get('testFilterMulti') == 'good_filter'
non_local['trigger_count'] += 1
except AssertionError as e:
sys.exit(e)
if non_local['trigger_count'] == 2:
sub_mgr.pubsub.greenlet.kill()
def publish_and_unsubscribe_handler(sub_id):
sub_mgr.publish('not_a_trigger', {'filterBoolean': False})
sub_mgr.publish('trigger_1', {'filterBoolean': True})
sub_mgr.publish('trigger_2', {'filterBoolean': True})
sub_mgr.pubsub.greenlet.join()
sub_mgr.unsubscribe(sub_id)
p1 = sub_mgr.subscribe(query, 'multiTrigger', callback,
{'filterBoolean': True,
'uga': 'UGA'}, {}, None, None)
p2 = p1.then(publish_and_unsubscribe_handler)
p2.get()
def validation_schema():
class Query(graphene.ObjectType):
placeholder = graphene.String()
class Subscription(graphene.ObjectType):
test_1 = graphene.String()
test_2 = graphene.String()
return graphene.Schema(query=Query, subscription=Subscription)
def schema(data):
class UserType(graphene.ObjectType):
id = graphene.String()
name = graphene.String()
class Query(graphene.ObjectType):
test_string = graphene.String()
class Subscription(graphene.ObjectType):
user = graphene.Field(UserType, id=graphene.String())
user_filtered = graphene.Field(UserType, id=graphene.String())
context = graphene.String()
error = graphene.String()
def resolve_user(self, info):
id = args['id']
name = data[args['id']]['name']
return UserType(id=id, name=name)
def resolve_user_filtered(self, name=name)
def resolve_context(self, info):
return context
def resolve_error(self, info):
raise Exception('E1')
return graphene.Schema(query=Query, subscription=Subscription)
def test_should_call_unsubscribe_when_client_closes_cxn(server_with_mocks):
node_script = '''
module.paths.push('{0}')
WebSocket = require('ws')
const SubscriptionClient =
require('subscriptions-transport-ws').SubscriptionClient
const client = new SubscriptionClient('ws://localhost:{1}/socket')
client.subscribe({{
query: `subscription useInfo($id: String) {{
user(id: $id) {{
id
name
}}
}}`,
operationName: 'useInfo',
variables: {{
id: 3,
}},
}},function (error,result) {{
// nothing
}}
)
setTimeout(() => {{
client.client.close()
}},500)
'''.format(
os.path.join(os.path.dirname(__file__), 'node_modules'), TEST_PORT)
try:
subprocess.check_output(
['node', '-e', node_script], stderr=subprocess.STDOUT, timeout=1)
except:
while True:
mock = server_with_mocks.get_Nowait()
if mock.name == 'on_unsubscribe':
mock.assert_called_once()
break
def test_should_trigger_on_unsubscribe_when_client_unsubscribes(
server_with_mocks):
node_script = '''
module.paths.push('{0}')
WebSocket = require('ws')
const SubscriptionClient =
require('subscriptions-transport-ws').SubscriptionClient
const client = new SubscriptionClient('ws://localhost:{1}/socket')
const subId = client.subscribe({{
query: `subscription useInfo($id: String) {{
user(id: $id) {{
id
name
}}
}}`,result) {{
// nothing
}})
client.unsubscribe(subId)
'''.format(
os.path.join(os.path.dirname(__file__), timeout=.2)
except:
while True:
mock = server_with_mocks.get_Nowait()
if mock.name == 'on_unsubscribe':
mock.assert_called_once()
break
def convert_date_to_string(type, registry=None):
return String(description=getattr(attribute, True)))
def convert_scalar_list_to_list(type, registry=None):
return List(String, description=attribute.attr_name)
def convert_list_to_list(type, description=attribute.attr_name)
def test_should_datetime_convert_string():
assert_attribute_conversion(UTCDateTimeAttribute(), graphene.String)
def test_should_binary_convert_string():
assert_attribute_conversion(BinaryAttribute(), graphene.String)
def test_should_char_convert_string():
assert_conversion(models.CharField, graphene.String)
def test_should_text_convert_string():
assert_conversion(models.TextField, graphene.String)
def test_should_slug_convert_string():
assert_conversion(models.SlugField, graphene.String)
def test_should_url_convert_string():
assert_conversion(models.URLField, graphene.String)
def test_should_ipaddress_convert_string():
assert_conversion(models.GenericIPAddressField, graphene.String)
def test_should_file_convert_string():
assert_conversion(models.FileField, graphene.String)
def test_should_image_convert_string():
assert_conversion(models.ImageField, graphene.String)
def test_should_time_convert_string():
assert_conversion(forms.TimeField, graphene.String)
def test_should_date_time_convert_string():
assert_conversion(forms.DateTimeField, graphene.String)
def test_should_char_convert_string():
assert_conversion(forms.CharField, graphene.String)
def test_should_email_convert_string():
assert_conversion(forms.EmailField, graphene.String)
def test_should_slug_convert_string():
assert_conversion(forms.SlugField, graphene.String)
def test_should_choice_convert_string():
assert_conversion(forms.ChoiceField, graphene.String)
def test_should_base_field_convert_string():
assert_conversion(forms.Field, graphene.String)
def test_should_regex_convert_string():
assert_conversion(forms.RegexField, graphene.String, '[0-9]+')
def convert_serializer_field_to_list_of_string(field):
return (graphene.List, graphene.String)
def test_should_char_convert_string():
assert_conversion(serializers.CharField, graphene.String)
def test_should_slug_convert_string():
assert_conversion(serializers.SlugField, graphene.String)
def test_should_url_convert_string():
assert_conversion(serializers.URLField, graphene.String)
def test_should_choice_convert_string():
assert_conversion(serializers.ChoiceField, choices=[])
def test_should_base_field_convert_string():
assert_conversion(serializers.Field, graphene.String)
def test_should_regex_convert_string():
assert_conversion(serializers.RegexField, regex='[0-9]+')
def test_should_duration_convert_string():
assert_conversion(serializers.DurationField, graphene.String)
def test_should_file_convert_string():
assert_conversion(serializers.FileField, graphene.String)
def test_should_filepath_convert_string():
assert_conversion(serializers.FilePathField, path='/')
def test_should_ip_convert_string():
assert_conversion(serializers.IPAddressField, graphene.String)
def test_should_image_convert_string():
assert_conversion(serializers.ImageField, graphene.String)
def convert_column_to_string(type, column, registry=None):
return graphene.String(
description=get_column_doc(column),
required=not(is_column_nullable(column)),
)
def convert_ndb_string_property(ndb_prop, registry=None):
return convert_ndb_scalar_property(String, ndb_prop)
def convert_ndb_key_propety(ndb_key_prop, registry=None):
"""
Two conventions for handling KeyProperties:
#1.
Given:
store_key = ndb.KeyProperty(...)
Result is 2 fields:
store_id = graphene.String() -> resolves to store_key.urlsafe()
store = NdbKeyField() -> resolves to entity
#2.
Given:
store = ndb.KeyProperty(...)
Result is 2 fields:
store_id = graphene.String() -> resolves to store_key.urlsafe()
store = NdbKeyField() -> resolves to entity
"""
is_repeated = ndb_key_prop._repeated
name = ndb_key_prop._code_name
if name.endswith('_key') or name.endswith('_keys'):
# Case #1 - name is of form 'store_key' or 'store_keys'
string_prop_name = rreplace(name, '_key', '_id', 1)
resolved_prop_name = name[:-4] if name.endswith('_key') else p.plural(name[:-5])
else:
# Case #2 - name is of form 'store'
singular_name = p.singular_noun(name) if p.singular_noun(name) else name
string_prop_name = singular_name + '_ids' if is_repeated else singular_name + '_id'
resolved_prop_name = name
return [
ConversionResult(name=string_prop_name, field=DynamicNdbKeyStringField(ndb_key_prop, registry=registry)),
ConversionResult(name=resolved_prop_name, field=DynamicNdbKeyReferenceField(ndb_key_prop, registry=registry))
]
def testGET_support_json_variables(self):
response = self.app.get('/graphql', params=dict(
query='query helloWho($who: String){ greet(who: $who) }',
variables=json.dumps({'who': "ekampf"})
))
response_dict = json.loads(response.body)
self.assertDictEqual(
response_dict.get('data'), {'greet': 'Hello ekampf!'}
)
def testPOST_support_json_variables(self):
response = self.app.post('/graphql', params=json.dumps(dict(
query='query helloWho($who: String){ greet(who: $who) }',
variables={'who': "ekampf"}
)))
response_dict = json.loads(response.body)
self.assertDictEqual(
response_dict.get('data'), {'greet': 'Hello ekampf!'}
)
def test_handles_poorly_formed_variables(self):
for method in (self.get, self.post):
response = method('/graphql', expect_errors=True, params=dict(
query='query helloWho($who: String){ greet(who: $who) }',
variables='who:You'
))
response_data = json.loads(response.body)
self.assertEqual(response.status_int, 400)
self.assertEqual(response_data['errors'][0]['message'], 'Variables are invalid JSON.')
def testStringProperty_shouldConvertToString(self):
self.__assert_conversion(ndb.StringProperty, graphene.String)
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。