Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class Meta:
model = EmployeeModel
interfaces = (relay.Node, )
class Role(SQLAlchemyObjectType):
class Meta:
model = RoleModel
interfaces = (relay.Node, )
class Query(graphene.ObjectType):
node = relay.Node.Field()
all_employees = SQLAlchemyConnectionField(Employee)
all_roles = SQLAlchemyConnectionField(Role)
role = graphene.Field(Role)
schema = graphene.Schema(query=Query, types=[Department, Employee, Role])
model = EmployeeModel
interfaces = (relay.Node, )
class Role(SQLAlchemyObjectType):
class Meta:
model = RoleModel
interfaces = (relay.Node, )
class Query(graphene.ObjectType):
node = relay.Node.Field()
all_employees = SQLAlchemyConnectionField(Employee)
all_roles = SQLAlchemyConnectionField(Role)
role = graphene.Field(Role)
schema = graphene.Schema(query=Query, types=[Department, Employee, Role])
class Role(SQLAlchemyObjectType):
class Meta:
model = RoleModel
interfaces = (relay.Node, )
class Query(graphene.ObjectType):
node = relay.Node.Field()
all_employees = SQLAlchemyConnectionField(Employee)
all_roles = SQLAlchemyConnectionField(Role)
role = graphene.Field(Role)
schema = graphene.Schema(query=Query, types=[Department, Employee, Role])
class Meta:
model = EmployeeModel
interfaces = (relay.Node, )
class Role(SQLAlchemyObjectType):
class Meta:
model = RoleModel
interfaces = (relay.Node, )
class Query(graphene.ObjectType):
node = relay.Node.Field()
all_employees = SQLAlchemyConnectionField(Employee)
all_roles = SQLAlchemyConnectionField(Role)
role = graphene.Field(Role)
schema = graphene.Schema(query=Query, types=[Department, Employee, Role])
def construct_fields(options):
only_fields = options.only_fields
exclude_fields = options.exclude_fields
inspected_model = sqlalchemyinspect(options.model)
fields = OrderedDict()
for name, column in inspected_model.columns.items():
is_not_in_only = only_fields and name not in only_fields
is_already_created = name in options.fields
is_excluded = name in exclude_fields or is_already_created
if is_not_in_only or is_excluded:
# We skip this field if we specify only_fields and is not
# in there. Or when we excldue this field in exclude_fields
continue
converted_column = convert_sqlalchemy_column(column, options.registry)
fields[name] = converted_column
# Get all the columns for the relationships on the model
for relationship in inspected_model.relationships:
is_not_in_only = only_fields and relationship.key not in only_fields
is_already_created = relationship.key in options.fields
is_excluded = relationship.key in exclude_fields or is_already_created
if is_not_in_only or is_excluded:
# We skip this field if we specify only_fields and is not
# in there. Or when we excldue this field in exclude_fields
continue
converted_relationship = convert_sqlalchemy_relationship(relationship, options.registry)
name = relationship.key
fields[name] = converted_relationship
return fields
# We skip this field if we specify only_fields and is not
# in there. Or when we excldue this field in exclude_fields
continue
converted_column = convert_sqlalchemy_column(column, options.registry)
fields[name] = converted_column
# Get all the columns for the relationships on the model
for relationship in inspected_model.relationships:
is_not_in_only = only_fields and relationship.key not in only_fields
is_already_created = relationship.key in options.fields
is_excluded = relationship.key in exclude_fields or is_already_created
if is_not_in_only or is_excluded:
# We skip this field if we specify only_fields and is not
# in there. Or when we excldue this field in exclude_fields
continue
converted_relationship = convert_sqlalchemy_relationship(relationship, options.registry)
name = relationship.key
fields[name] = converted_relationship
return fields
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 is_mapped(options.model), (
'You need to pass a valid SQLAlchemy Model in '
'{}.Meta, received "{}".'
).format(name, options.model)
cls = ObjectTypeMeta.__new__(cls, name, bases, dict(attrs, _meta=options))
options.registry.register(cls)
options.sqlalchemy_fields = yank_fields_from_attrs(
construct_fields(options),
_as=Field,
def get_global_registry():
global registry
if not registry:
registry = Registry()
return registry
options.sqlalchemy_fields = yank_fields_from_attrs(
construct_fields(options),
_as=Field,
)
options.fields = merge(
options.interface_fields,
options.sqlalchemy_fields,
options.base_fields,
options.local_fields
)
return cls
class SQLAlchemyObjectType(six.with_metaclass(SQLAlchemyObjectTypeMeta, ObjectType)):
@classmethod
def is_type_of(cls, root, context, info):
if isinstance(root, cls):
return True
if not is_mapped(type(root)):
raise Exception((
'Received incompatible instance "{}".'
).format(root))
return type(root) == cls._meta.model
@classmethod
def get_node(cls, id, context, info):
try:
model = cls._meta.model
query = get_query(model, context)
return query.get(id)
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, SQLAlchemyObjectTypeMeta):
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: