python类Field()的实例源码

def test_filter_filterset_information_on_meta_related():
    class ReporterFilterNode(DjangoObjectType):

        class Meta:
            model = Reporter
            interfaces = (Node, )
            filter_fields = ['first_name', 'articles']

    class ArticleFilterNode(DjangoObjectType):

        class Meta:
            model = Article
            interfaces = (Node, )
            filter_fields = ['headline', 'reporter']

    class Query(ObjectType):
        all_reporters = DjangoFilterConnectionField(ReporterFilterNode)
        all_articles = DjangoFilterConnectionField(ArticleFilterNode)
        reporter = Field(ReporterFilterNode)
        article = Field(ArticleFilterNode)

    schema = Schema(query=Query)
    articles_field = ReporterFilterNode._meta.fields['articles'].get_type()
    assert_arguments(articles_field, 'headline', 'reporter')
    assert_not_orderable(articles_field)

def convert_local_structured_property(ndb_structured_property, registry=None):
    is_required = ndb_structured_property._required
    is_repeated = ndb_structured_property._repeated
    model = ndb_structured_property._modelclass
    name = ndb_structured_property._code_name

    def dynamic_type():
        _type = registry.get_type_for_model(model)
        if not _type:
            return None

        if is_repeated:
            _type = List(_type)

        if is_required:
            _type = NonNull(_type)

        return Field(_type)

    field = Dynamic(dynamic_type)
    return ConversionResult(name=name, field=field)

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, args, context, info):
            id = args['id']
            name = data[args['id']]['name']
            return UserType(id=id, name=name)

        def resolve_user_filtered(self, args, context, info):
            id = args['id']
            name = data[args['id']]['name']
            return UserType(id=id, name=name)

        def resolve_context(self, args, context, info):
            return context

        def resolve_error(self, args, context, info):
            raise Exception('E1')

    return graphene.Schema(query=Query, subscription=Subscription)

def convert_relationship_to_dynamic(type, attribute, registry=None):
    def dynamic_type():
        _type = registry.get_type_for_model(attribute.model)
        if not _type:
            return None

        if isinstance(attribute, OneToOne):
            return Field(_type)

        if isinstance(attribute, OneToMany):
            if is_node(_type):
                return PynamoConnectionField(_type)
            return Field(List(_type))

    return Dynamic(dynamic_type)

def assert_attribute_conversion(attribute, graphene_field, **kwargs):
    graphene_type = convert_pynamo_attribute(attribute, attribute)
    assert isinstance(graphene_type, graphene_field)
    field = graphene_type.Field()
    return field

def test_should_onetoone_convert_field():
    class A(PynamoObjectType):
        class Meta:
            model = Article
            interfaces = (Node,)

    dynamic_field = convert_pynamo_attribute(Reporter.favorite_article, Reporter.favorite_article, A._meta.registry)
    assert isinstance(dynamic_field, Dynamic)
    graphene_type = dynamic_field.get_type()
    assert isinstance(graphene_type, graphene.Field)
    assert graphene_type.type == A

def test_should_onetomany_convert_nonnode_field():
    class A(PynamoObjectType):
        class Meta:
            model = Article

    dynamic_field = convert_pynamo_attribute(Reporter.articles, Reporter.articles, A._meta.registry)
    assert isinstance(dynamic_field, Dynamic)
    graphene_type = dynamic_field.get_type()
    assert isinstance(graphene_type, graphene.Field)
    assert graphene_type.type == graphene.List(A)

def test_root_scan_should_warn_on_params():
    class ArticleNode(PynamoObjectType):
        class Meta:
            model = Article
            interfaces = (Node,)

    class Query(graphene.ObjectType):
        node = Node.Field()
        articles = PynamoConnectionField(ArticleNode)

    query = '''
        query ArticlesQuery {
          articles(after: "QXJ0aWNsZU5vZGU6MQ==") {
            edges {
              node {
                id
                headline
              }
            }
          }
        }
    '''

    schema = graphene.Schema(query=Query)
    result = schema.execute(query)
    assert result.errors
    assert isinstance(result.errors[0].original_error, NotImplementedError)

def test_node_replacedfield():
    idfield = Human._meta.fields['pub_date']
    assert isinstance(idfield, Field)
    assert idfield.type == Int

def to_attrs(self):
        return {
            'single': self.single(),
            'list': self.list(),
            'fields': self.fields,
            'model': self.model,
            'mutate': self.mutate().Field()
        }

def resolver(g_schema, mongo_doc, operators_single=None, operators_list=None, is_list=False, validator=None):
        def auto_resolver(root, args, contex, info):
            return resolver_query(g_schema, mongo_doc, args, info, is_list=is_list, validator=validator)

        if is_list:
            return graphene.List(g_schema, **operators_list, resolver=auto_resolver)
        else:
            return graphene.Field(g_schema, **operators_single, resolver=auto_resolver)

def reference_field(f_name, mongo_field):
    """ Generate a schema for RefereceField, or get a schema already done saved in _generated_schemas """
    from graphene_mongo import MongoSchema

    document = mongo_field.document_type_obj  # document that this ReferenceField references

    schema = MongoSchema.get_or_generate_schema(document).schema
    return graphene.Field(schema)

def assert_orderable(field):
    args = get_args(field)
    assert 'order_by' in args, \
        'Field cannot be ordered'

def assert_conversion(django_field, graphene_field, *args, **kwargs):
    field = django_field(help_text='Custom Help Text', null=True, *args, **kwargs)
    graphene_type = convert_django_field(field)
    assert isinstance(graphene_type, graphene_field)
    field = graphene_type.Field()
    assert field.description == 'Custom Help Text'
    nonnull_field = django_field(null=False, *args, **kwargs)
    if not nonnull_field.null:
        nonnull_graphene_type = convert_django_field(nonnull_field)
        nonnull_field = nonnull_graphene_type.Field()
        assert isinstance(nonnull_field.type, graphene.NonNull)
        return nonnull_field
    return field

def test_should_manytomany_convert_connectionorlist_list():
    class A(DjangoObjectType):

        class Meta:
            model = Reporter

    graphene_field = convert_django_field(Reporter._meta.local_many_to_many[0], A._meta.registry)
    assert isinstance(graphene_field, graphene.Dynamic)
    dynamic_field = graphene_field.get_type()
    assert isinstance(dynamic_field, graphene.Field)
    assert isinstance(dynamic_field.type, graphene.List)
    assert dynamic_field.type.of_type == A

def test_should_onetoone_reverse_convert_model():
    # Django 1.9 uses 'rel', <1.9 uses 'related
    related = getattr(Film.details, 'rel', None) or \
        getattr(Film.details, 'related')

    class A(DjangoObjectType):

        class Meta:
            model = FilmDetails

    graphene_field = convert_django_field(related, A._meta.registry)
    assert isinstance(graphene_field, graphene.Dynamic)
    dynamic_field = graphene_field.get_type()
    assert isinstance(dynamic_field, graphene.Field)
    assert dynamic_field.type == A

def test_should_query_simplelazy_objects():
    class ReporterType(DjangoObjectType):

        class Meta:
            model = Reporter
            only_fields = ('id', )

    class Query(graphene.ObjectType):
        reporter = graphene.Field(ReporterType)

        def resolve_reporter(self, info):
            return SimpleLazyObject(lambda: Reporter(id=1))

    schema = graphene.Schema(query=Query)
    query = '''
        query {
          reporter {
            id
          }
        }
    '''
    result = schema.execute(query)
    assert not result.errors
    assert result.data == {
        'reporter': {
            'id': '1'
        }
    }

def test_should_query_well():
    class ReporterType(DjangoObjectType):

        class Meta:
            model = Reporter

    class Query(graphene.ObjectType):
        reporter = graphene.Field(ReporterType)

        def resolve_reporter(self, info):
            return Reporter(first_name='ABA', last_name='X')

    query = '''
        query ReporterQuery {
          reporter {
            firstName,
            lastName,
            email
          }
        }
    '''
    expected = {
        'reporter': {
            'firstName': 'ABA',
            'lastName': 'X',
            'email': ''
        }
    }
    schema = graphene.Schema(query=Query)
    result = schema.execute(query)
    assert not result.errors
    assert result.data == expected

def test_should_handle_inherited_choices():
    class BaseModel(models.Model):
        choice_field = models.IntegerField(choices=((0, 'zero'), (1, 'one')))

    class ChildModel(BaseModel):
        class Meta:
            proxy = True

    class BaseType(DjangoObjectType):
        class Meta:
            model = BaseModel

    class ChildType(DjangoObjectType):
        class Meta:
            model = ChildModel

    class Query(graphene.ObjectType):
        base = graphene.Field(BaseType)
        child = graphene.Field(ChildType)

    schema = graphene.Schema(query=Query)
    query = '''
        query {
          child {
            choiceField
          }
        }
    '''
    result = schema.execute(query)
    assert not result.errors

def test_should_base_field_convert_string():
    assert_conversion(serializers.Field, graphene.String)

def test_should_model_convert_field():

    class MyModelSerializer(serializers.ModelSerializer):
        class Meta:
            model = None
            fields = '__all__'

    assert_conversion(MyModelSerializer, graphene.Field, is_input=False)

def convert_ndb_scalar_property(graphene_type, ndb_prop, registry=None, **kwargs):
    kwargs['description'] = "%s %s property" % (ndb_prop._name, graphene_type)
    _type = graphene_type

    if ndb_prop._repeated:
        _type = List(_type)

    if ndb_prop._required:
        _type = NonNull(_type)

    return Field(_type, **kwargs)

def convert_ndb_json_property(ndb_prop, registry=None):
    return Field(JSONString, description=ndb_prop._name)

def convert_ndb_datetime_property(ndb_prop, registry=None):
    return Field(DateTime, description=ndb_prop._name)

def make_resolvable_fields(inner_fields):
    result = {}

    for name, field in inner_fields.items():
        result.update({
            name: Field(String, **get_field_args(field)),
            'resolve_{}'.format(name): partial(resolve, name),
        })
    return result

def convert_fields_to_graph(fields):
    for name, field in fields.items():
        inner_fields = get_fields(field, callables=True)
        inner_fields = make_resolvable_fields(inner_fields)

        field_class = type(name.capitalize(), (ObjectType, ), inner_fields)
        fields[name] = Field(field_class)

    return fields

def __new__(cls, name, bases, attrs):
        # Also ensure initialization is only performed for subclasses of Model
        # (excluding Model class itself).
        if not is_base_type(bases, PynamoObjectTypeMeta):
            return type.__new__(cls, name, bases, attrs)

        options = Options(
            attrs.pop('Meta', None),
            name=name,
            description=attrs.pop('__doc__', None),
            model=None,
            local_fields=None,
            only_fields=(),
            exclude_fields=(),
            id='id',
            interfaces=(),
            registry=None
        )

        if not options.registry:
            options.registry = get_global_registry()
        assert isinstance(options.registry, Registry), (
            'The attribute registry in {}.Meta needs to be an'
            ' instance of Registry, received "{}".'
        ).format(name, options.registry)

        assert (inspect.isclass(options.model) and issubclass(options.model, Model)), (
            'You need to pass a valid PynamoDB Model in '
            '{}.Meta, received "{}".'
        ).format(name, options.model)

        cls = ObjectTypeMeta.__new__(cls, name, bases, dict(attrs, _meta=options))

        options.registry.register(cls)

        options.pynamo_fields = yank_fields_from_attrs(
            construct_fields(options),
            _as=Field,
        )
        options.fields = merge(
            options.interface_fields,
            options.pynamo_fields,
            options.base_fields,
            options.local_fields
        )

        return cls

def test_should_custom_identifier():
    class EditorNode(PynamoObjectType):
        class Meta:
            model = Editor
            interfaces = (Node,)

    class Query(graphene.ObjectType):
        node = Node.Field()
        all_editors = PynamoConnectionField(EditorNode)

    query = '''
        query EditorQuery {
          allEditors {
            edges {
                node {
                    id,
                    name
                }
            }
          },
          node(id: "RWRpdG9yTm9kZTox") {
            ...on EditorNode {
              name
            }
          }
        }
    '''
    expected = {
        'allEditors': {
            'edges': [{
                'node': {
                    'id': 'RWRpdG9yTm9kZTox',
                    'name': 'John'
                }
            }]
        },
        'node': {
            'name': 'John'
        }
    }

    schema = graphene.Schema(query=Query)
    result = schema.execute(query)
    assert not result.errors
    assert result.data['allEditors'] == expected['allEditors']

def test_should_return_empty_cursors_on_empty():
    class ArticleNode(PynamoObjectType):
        class Meta:
            model = Article
            interfaces = (Node,)

    class ReporterNode(PynamoObjectType):
        class Meta:
            model = Reporter
            interfaces = (Node,)

    class Query(graphene.ObjectType):
        node = Node.Field()
        reporter = graphene.Field(ReporterNode)

        def resolve_reporter(self, *args, **kwargs):
            return Reporter.get(2)

    query = '''
        query ReporterQuery {
          reporter {
            id,
            firstName,
            articles(first: 1) {
              edges {
                node {
                  id
                  headline
                }
              }
              pageInfo {
                hasNextPage
                hasPreviousPage
                startCursor
                endCursor
              }
            }
            lastName,
            email
          }
        }
    '''
    expected = {
        'reporter': {
            'articles': {
                'edges': [],
                'pageInfo': {
                    'hasNextPage': False,
                    'hasPreviousPage': False,
                    'startCursor': '',
                    'endCursor': ''
                }
            }
        }
    }

    schema = graphene.Schema(query=Query)
    result = schema.execute(query)
    assert not result.errors
    assert result.data['reporter']['articles']['edges'] == expected['reporter']['articles']['edges']
    assert result.data['reporter']['articles']['pageInfo'] == expected['reporter']['articles']['pageInfo']

def test_should_support_last():
    class ArticleNode(PynamoObjectType):
        class Meta:
            model = Article
            interfaces = (Node,)

    class ReporterNode(PynamoObjectType):
        class Meta:
            model = Reporter
            interfaces = (Node,)

    class Query(graphene.ObjectType):
        node = Node.Field()
        reporter = graphene.Field(ReporterNode)

        def resolve_reporter(self, *args, **kwargs):
            return Reporter.get(1)

    query = '''
        query ReporterQuery {
          reporter {
            id,
            firstName,
            articles(last: 1) {
              edges {
                node {
                  id
                  headline
                }
              }
            }
            lastName,
            email
          }
        }
    '''
    expected = {
        'reporter': {
            'articles': {
                'edges': [{
                    'node': {
                        'id': 'QXJ0aWNsZU5vZGU6Mw==',
                        'headline': 'My Article'
                    }
                }]
            }
        }
    }

    schema = graphene.Schema(query=Query)
    result = schema.execute(query)
    assert not result.errors
    assert result.data['reporter']['articles']['edges'] == expected['reporter']['articles']['edges']