I have the following Python module that implements Graphene.
import graphene
import json
import psycopg2
connection = psycopg2.connect(user='postgres', password='xxxxx', host='127.0.0.1', port='5432', database='xxxx')
cursor = connection.cursor()
class Entity(graphene.ObjectType):
name = graphene.String()
class Query(graphene.ObjectType):
entity_relationships = graphene.List(Entity, entity=graphene.String())
postgresql_version = graphene.String
path = graphene.String(name=graphene.String(), path=graphene.String(), accumulate=graphene.String(), depth=graphene.Int())
def get_foreign_relationships(self, entity):
cursor.execute('''
SELECT
tc.table_schema, tc.constraint_name, tc.table_name, kcu.column_name, ccu.table_schema AS foreign_table_schema, ccu.table_name AS foreign_table_name, ccu.column_name AS foreign_column_name
FROM information_schema.table_constraints AS tc
JOIN information_schema.key_column_usage AS kcu ON tc.constraint_name = kcu.constraint_name
AND tc.table_schema = kcu.table_schema
JOIN information_schema.constraint_column_usage AS ccu ON ccu.constraint_name = tc.constraint_name
AND ccu.table_schema = tc.table_schema
WHERE tc.constraint_type = 'FOREIGN KEY' AND tc.table_name='{}';'''.format(entity))
result = cursor.fetchall()
result_array = []
for record in result:
new_entity = Entity(name=record[5])
result_array.append(new_entity)
return result_array
def is_relationship(self, referencing, referenced):
foreign_relationships = self.get_foreign_relationships(referencing)
if referenced in list(map(lambda table: relationship[5], foreign_relationships)):
return True
else:
return False
def traverse(self, entities, direction):
for i in range(len(entities)):
if i > 0 and i < len(entities)-1:
if not self.is_relationship(entities[i], entities[i-1]):
raise "entity {} has no relation with entity {}".format(entities[i], entities[i-1])
return True
def validate_path(self, path):
entities = path.split('/')
self.traverse(entities)
return True
def resolve_path(self, info, name, path, accumulate, depth):
if self.validate_path(path):
return "Éste método regresará los datos anidados para el camino '{}'".format(path)
def resolve_entity_relationships(self, info, entity):
result_array = self.get_foreign_relationships(entity)
return result_array
def resolve_postgresql_version(self, info):
cursor.execute("SELECT version();")
record = cursor.fetchone()
return record
def execute_query(query_to_execute):
queries = {
'postgresqlVersion': '''
{
postgresqlVersion
}
''',
'entityRelationships': '''
{
entityRelationships (entity: "inventory_productitem") {
name
}
}
''',
'path': '''
{
path(name: "Ventas", path: "inventory_family/inventory_product/inventory_sale", accumulate: "_calculate(amount*price)", depth: 1)
}
'''
}
schema = graphene.Schema(query=Query)
result = schema.execute(queries[query_to_execute])
dict_result = dict(result.data.items())
print(json.dumps(dict_result, indent=2))
execute_query('path')
The problem is that resolvers that call other methods of the class raise this error:
Traceback (most recent call last):
File "/Users/hugovillalobos/Documents/Code/agrowareproject/backend/AgrowareVenv/lib/python3.7/site-packages/graphql/execution/executor.py", line 452, in resolve_or_error
return executor.execute(resolve_fn, source, info, **args)
File "/Users/hugovillalobos/Documents/Code/agrowareproject/backend/AgrowareVenv/lib/python3.7/site-packages/graphql/execution/executors/sync.py", line 16, in execute
return fn(*args, **kwargs)
File "query.py", line 59, in resolve_path
if self.validate_path(path):
graphql.error.located_error.GraphQLLocatedError: 'NoneType' object has no attribute 'validate_path'
The line that rises the error is:
if self.validate_path(path):
I don't know why, if method validate_path()
is in the same class that the method that calls it.
This is because Graphene considering all resolver methods are staticmethods
From the doc,
Sometimes this argument will be named
self
in Graphene code, but this can be misleading due to Implicit staticmethod while executing queries in Graphene.
So, you must rearrange your code as follows,
def validate_path(path):
# do something
return True # ot False depends on your evaluation
class Query(graphene.ObjectType):
path = graphene.String(
name=graphene.String(),
path=graphene.String(),
accumulate=graphene.String(),
depth=graphene.Int()
)
@staticmethod
def resolve_path(parent, info, name, path, accumulate, depth):
if validate_path(path):
return "Validated path"
return "Path Not Valid"