Graphene-Python：从Django模型自动生成模式 [英] Graphene-Python: automatic schema generation from Django model
问题描述
我正在尝试从Django模型生成Graphene模式。我正在尝试通过依次遍历应用程序和模型，然后将适当的属性添加到生成的架构来做到这一点。
I am trying to generate a Graphene schema from a Django model. I am trying to do this by iterating through the apps then the models and then adding the appropriate attributes to the generated schema.
这是代码：
registry = {}
def register(target_class):
registry[target_class.__name__] = target_class
def c2u(name):
s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name)
return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower()
def s2p(name):
s1 = re.sub("y$", "ie", name)
return "{}s".format(s1)
class AutoSchemaMeta(type):
def __new__(meta, clsname, superclasses, attributedict):
new_class = type(clsname, superclasses, attributedict)
for app_name in new_class.app_models.split(","):
app_models = apps.get_app_config(app_name.strip()).get_models()
for model in app_models:
model_name = model._meta.model_name
_model_name = c2u(model_name)
if hasattr(new_class,_model_name):
continue
_node_class = type("{}Node".format(model_name.title()),
(DjangoObjectType,),
{"Meta":{"model": model, "interfaces": (Node,), "filter_fields": []}})
register(_node_class)
setattr(new_class, "all_{}".format(s2p(_model_name)), DjangoFilterConnectionField(_node_class))
setattr(new_class, _model_name, Node.Field(_node_class))
print(new_class.__dict__)
return new_class
class Query(metaclass=AutoSchemaMeta):
app_models = "app1,app2"
何时我运行我的应用程序时遇到异常：
When I run my application I get an exception:
AssertionError：在
模式中发现了具有相同名称的不同类型：WorkflowNode，
AssertionError: Found different types with the same name in the schema: WorkflowNode, WorkflowNode.
结果是有一个已经定义为WorkflowNode的类，我不想重写它。因此，现在我要坚持找出已经定义的类。
Turns out there is a class already defined as WorkflowNode and I do not want to override it. So now I am stuck at finding out the classes that are already defined.
如果hasattr（new_class，_model_name）：继续，我已经按的属性名称进行了排除：不依赖约定并找出在其他地方定义的所有 Node 类，如果存在，则使用它们代替我自动创建的类</ p>
I am already excluding by attributes name with if hasattr(new_class,_model_name): continue but I would like to not rely on conventions and find out also all Nodeclasses that have been defined elsewhere and if they exist use them instead of the one I am creating automatically
推荐答案
我尝试了提出的解决方案，但是由于很多原因，该方法不起作用，包括与graphene.ObjectType的元类冲突，因此我创建了一个效果很好的解决方案好：
I tried the proposed solution but it doesn't work for a lot of reasons, including metaclass conflicts with graphene.ObjectType so I created a solution that works pretty well:
，您将为子类化的ObjectType提供ORM模型的列表（在我的情况下为SQLAlchemy），它会自动创建模式。唯一要做的就是如果需要为任何字段添加额外的过滤选项，则添加特殊处理。
you would provide the subclassed ObjectType a list of your ORM models (in my case SQLAlchemy) and it auto creates the schema. The only thing left to do would be to add special handling if you needed to add extra filtering options for any of the fields.
class SQLAlchemyAutoSchemaFactory(graphene.ObjectType):
@staticmethod
def set_fields_and_attrs(klazz, node_model, field_dict):
_name = camel_to_snake(node_model.__name__)
field_dict[f'all_{(s2p(_name))}'] = FilteredConnectionField(node_model)
field_dict[_name] = node_model.Field()
# log.info(f'interface:{node_model.__name__}')
setattr(klazz, _name, node_model.Field())
setattr(klazz, "all_{}".format(s2p(_name)), FilteredConnectionField(node_model))
@classmethod
def __init_subclass_with_meta__(
cls,
interfaces=(),
models=(),
excluded_models=(),
default_resolver=None,
_meta=None,
**options
):
if not _meta:
_meta = ObjectTypeOptions(cls)
fields = OrderedDict()
for interface in interfaces:
if issubclass(interface, SQLAlchemyInterface):
SQLAlchemyAutoSchemaFactory.set_fields_and_attrs(cls, interface, fields)
for model in excluded_models:
if model in models:
models = models[:models.index(model)] + models[models.index(model) + 1:]
possible_types = ()
for model in models:
model_name = model.__name__
_model_name = camel_to_snake(model.__name__)
if hasattr(cls, _model_name):
continue
if hasattr(cls, "all_{}".format(s2p(_model_name))):
continue
for iface in interfaces:
if issubclass(model, iface._meta.model):
model_interface = (iface,)
break
else:
model_interface = (CustomNode,)
_node_class = type(model_name,
(SQLAlchemyObjectType,),
{"Meta": {"model": model, "interfaces": model_interface, "only_fields": []}})
fields["all_{}".format(s2p(_model_name))] = FilteredConnectionField(_node_class)
setattr(cls, "all_{}".format(s2p(_model_name)), FilteredConnectionField(_node_class))
fields[_model_name] = CustomNode.Field(_node_class)
setattr(cls, _model_name, CustomNode.Field(_node_class))
possible_types += (_node_class,)
if _meta.fields:
_meta.fields.update(fields)
else:
_meta.fields = fields
_meta.schema_types = possible_types
super(SQLAlchemyAutoSchemaFactory, cls).__init_subclass_with_meta__(_meta=_meta, default_resolver=default_resolver, **options)
@classmethod
def resolve_with_filters(cls, info: ResolveInfo, model: Type[SQLAlchemyObjectType], **kwargs):
query = model.get_query(info)
for filter_name, filter_value in kwargs.items():
model_filter_column = getattr(model._meta.model, filter_name, None)
if not model_filter_column:
continue
if isinstance(filter_value, SQLAlchemyInputObjectType):
filter_model = filter_value.sqla_model
q = FilteredConnectionField.get_query(filter_model, info, sort=None, **kwargs)
# noinspection PyArgumentList
query = query.filter(model_filter_column == q.filter_by(**filter_value))
else:
query = query.filter(model_filter_column == filter_value)
return query
，您就可以这样创建查询：
and you create the Query like this:
class Query(SQLAlchemyAutoSchemaFactory):
class Meta:
interfaces = (Interface1, Interface2,)
models = (*entities_for_iface1, *entities_for_iface2, *other_entities,)
excluded_models = (base_model_for_iface1, base_model_for_iface2)
创建这样的接口：
class Interface1(SQLAlchemyInterface):
class Meta:
name = 'Iface1Node'
model = Iface1Model
和SQLAlchemyInterface：
and SQLAlchemyInterface:
class SQLAlchemyInterface(Node):
@classmethod
def __init_subclass_with_meta__(
cls,
model=None,
registry=None,
only_fields=(),
exclude_fields=(),
connection_field_factory=default_connection_field_factory,
_meta=None,
**options
):
_meta = SQLAlchemyInterfaceOptions(cls)
_meta.name = f'{cls.__name__}Node'
autoexclude_columns = exclude_autogenerated_sqla_columns(model=model)
exclude_fields += autoexclude_columns
assert is_mapped_class(model), (
"You need to pass a valid SQLAlchemy Model in " '{}.Meta, received "{}".'
).format(cls.__name__, model)
if not registry:
registry = get_global_registry()
assert isinstance(registry, Registry), (
"The attribute registry in {} needs to be an instance of "
'Registry, received "{}".'
).format(cls.__name__, registry)
sqla_fields = yank_fields_from_attrs(
construct_fields(
model=model,
registry=registry,
only_fields=only_fields,
exclude_fields=exclude_fields,
connection_field_factory=connection_field_factory
),
_as=Field
)
if not _meta:
_meta = SQLAlchemyInterfaceOptions(cls)
_meta.model = model
_meta.registry = registry
connection = Connection.create_type(
"{}Connection".format(cls.__name__), node=cls)
assert issubclass(connection, Connection), (
"The connection must be a Connection. Received {}"
).format(connection.__name__)
_meta.connection = connection
if _meta.fields:
_meta.fields.update(sqla_fields)
else:
_meta.fields = sqla_fields
super(SQLAlchemyInterface, cls).__init_subclass_with_meta__(_meta=_meta, **options)
@classmethod
def Field(cls, *args, **kwargs): # noqa: N802
return NodeField(cls, *args, **kwargs)
@classmethod
def node_resolver(cls, only_type, root, info, id):
return cls.get_node_from_global_id(info, id, only_type=only_type)
@classmethod
def get_node_from_global_id(cls, info, global_id, only_type=None):
try:
node: DeclarativeMeta = one_or_none(session=info.context.get('session'), model=cls._meta.model, id=global_id)
return node
except Exception:
return None
@classmethod
def from_global_id(cls, global_id):
return global_id
@classmethod
def to_global_id(cls, type, id):
return id
@classmethod
def resolve_type(cls, instance, info):
if isinstance(instance, graphene.ObjectType):
return type(instance)
graphene_model = get_global_registry().get_type_for_model(type(instance))
if graphene_model:
return graphene_model
else:
raise ValueError(f'{instance} must be a SQLAlchemy model or graphene.ObjectType')
这篇关于Graphene-Python：从Django模型自动生成模式的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
