在运行时石墨烯上创建动态模式

Question

我几乎花了 3 天时间才找到一种在 python 石墨烯中创建动态模式的方法。我能找到的唯一相关结果是下面的 link： https://github.com/graphql-python/graphene/blob/master/graphene/types/dynamic.py 但是我找不到它的任何文档。

整个想法是创建一个动态模式。我想提供一个兼容 GraphQL 的 API，即使模型没有在代码中定义，用户也可以查询我的内容。换句话说，我想即时创建模型。我不知道该怎么办。

如能提供示例，将不胜感激。

更新：

我的项目是一个 Headless CMS，它具有用户可以创建自己的内容类型的功能，我想提供一个 GraphQL 接口，使一切变得更简单、更灵活。

这是我在数据库中的内容类型示例：

{
  "id": "author",
  "name": "Book Author",
  "desc": "",
  "options":[
    {
      "id": "author_faname",
      "label": "Sample Sample",
      "type": "text",
      "required": true,
      "placeholder":"One Two Three Four"
    },
    {
      "id": "author_enname",
      "label": "Sample label",
      "type": "text",
      "required": true,
      "placeholder":"Sample Placeholder"
    }
  ]
}

这是基于该内容类型在数据库中存储的内容：

{
  "id": "9rqgbrox10",
  "content_type": "author",
  "data":{
    "author_fname":"Jimmy",
    "author_ename":"Hello"
  }
}

现在，由于我的模型没有在代码中声明，它们完全在数据库中，我想即时制作我的模式，但我不知道什么是最好的解决方案。我知道应该有一种方法，因为其他 Headless CMS 项目正在提供这种方法。

提前致谢！

Answer 1

基本上，架构是这样创建的：

class MyType(graphene.ObjectType):
    something = graphene.String()

class Query(graphene.ObjectType):
    value = graphene.Field(MyType)

schema = graphene.Schema(query=Query, types=[MyType])

首先，为了添加某种动态，您很可能希望将上述代码包装在函数中，例如 create_schema()。

然后，当你想在运行时动态创建一个class，上面的代码可以改写成这样：

def create_schema():
    MyType = type('MyType', (graphene.ObjectType,), {
        'something': graphene.String(),
    })

    Query = type('Query', (graphene.ObjectType,), {
        'value': graphene.Field(MyType),
    })

    return graphene.Schema(query=Query, types=[MyType])

对于您的示例，它可能看起来像这样：

def make_resolver(record_name, record_cls):
    def resolver(self, info):
        data = ...
        return record_cls(...)
    resolver.__name__ = 'resolve_%s' % record_name
    return resolver

def create_schema(db):
    record_schemas = {}
    for record_type in db.get_record_types():
        classname = record_type['id'].title()  # 'Author'
        fields = {}
        for option in record_type['options']:
            field_type = {
                'text': graphene.String,
                ...
            }[option['type']
            fields[option['id']] = field_type()  # maybe add label as description?
        rec_cls = type(
            classname,
            (graphene.ObjectType,), 
            fields,
            name=record_type['name'],
            description=record_type['desc'],
        )
        record_schemas[record_type['id']] = rec_cls

    # create Query in similar way
    fields = {}
    for key, rec in record_schemas:
        fields[key] = graphene.Field(rec)
        fields['resolve_%s' % key] = make_resolver(key, rec)
    Query = type('Query', (graphene.ObjectType,), fields)

    return graphene.Schema(query=Query, types=list(record_schemas.values()))

请注意，如果您尝试将新字段插入现有 class，像这样 - MyType.another_field = graphene.String(), 那么它将不起作用：那是因为当 graphene.ObjectType class 被实例化时，它的所有字段都记录在 self._meta.fields OrderedDict 中。并且更新它并不像 MyType._meta.fields['another_field'] = thefield 那样简单-有关详细信息，请参阅 graphene.ObjectType.__init_subclass_with_meta__ 的代码。

因此，如果您的架构是动态更改的，那么从头开始完全重新创建它可能比修补它更好。

Answer 2

我想分享另一种巧妙的方法。

因此，问题在于 graphene.ObjectType 不是常规 Python class。它有一个特殊的metaclass that you can see implemented here。目前 Python 负责继承过程（当 class 本身被初始化时），石墨烯会做一些操作来注册类型。继承发生后，我没有找到更改类型的方法。但是，如果您只想从预定义的样板文件（如我）或其他来源生成模式，您可以这样做。我首先定义一个方便的继承方法：

def inherit_from(Child, Parent, persist_meta=False):
    """Return a class that is equivalent to Child(Parent) including Parent bases."""
    PersistMeta = copy(Child.Meta) if hasattr(Child, 'Meta') else None

    if persist_meta:
        Child.Meta = PersistMeta

    # Prepare bases
    child_bases = inspect.getmro(Child)
    parent_bases = inspect.getmro(Parent)
    bases = tuple([item for item in parent_bases if item not in child_bases]) + child_bases

    # Construct the new return type
    try:
        Child = type(Child.__name__, bases, Child.__dict__.copy())
    except AttributeError as e:
        if str(e) == 'Meta':
            raise AttributeError('Attribute Error in graphene library. Try setting persist_meta=True in the inherit_from method call.')
        raise e

    if persist_meta:
        Child.Meta = PersistMeta

    return Child

现在的关键是在类型的 class 不再更改时执行继承。

def context_resolver_factory(attr):
    """Create a simple resolver method with default return value None."""

    def resolver(obj, info):
        return info.context.get(attr, None)

    return resolver


class User:
    id = graphene.ID()
    name = graphene.String(resolver=lambda user, info: user.name)


class Query: pass
    me = graphene.Field(User)

    def resolve_me(self, info):
        return info.context["user"]


inherit_from(User, graphene.ObjectType)  # no changes to User class are possible after this line

# method 1: sometimes it's really neat and clean to include a resolver in the field definition
setattr(Query, 'user', graphene.User(resolver=context_resolver_factory('user'))
# or even use lambda if a factory is still overkill
setattr(Query, 'user', graphene.User(resolver=lambda query, info: info.context["user"]))


# method 2: if you want to set the resolver separately, you can do it this way
setattr(Query, 'user', graphene.User())
setattr(Query, 'resolve_user', context_resolver_factory('user'))

# any changes to `Query.Meta` can be done here too

inherit_from(Query, graphene.ObjectType)  # no changes to Query class are possible after this line

schema = graphene.Schema(query=Query)

我的小图书馆从样板 class 生成所有内容，如下所示：

@register_type('Product')
class ProductType:
    class Meta:
        model = Product
        fields = '__all__'
        related_fields = {
            NestedField('tags', TagType),
            NestedField('related_products', 'self'),
        }
        lookups = {
            'id': graphene.ID(),
            'name': graphene.String(description="Name"),
            'ean': graphene.String(),
            'brand': graphene.String()
        }
        filters = {
            'ids': IDFilter,
            'django_filter': DjangoFilter,
            'pagination': PaginationFilter,
            'search_name': ProductMLNSearchFilter
        }

最大的挑战是 NestedFields 和在收到请求时找出自动 Django ORM 查询 select/prefetch，但除非相关，否则我不会详细介绍。

在运行时石墨烯上创建动态模式

Creating Dynamic Schema on Runtime Graphene

python

graphql

graphene-python