Adding support for sqlalchemy

This commit is contained in:
Jeremy Feinstein 2015-11-21 15:21:40 -05:00
parent b1e0c3b533
commit 487543206c
7 changed files with 259 additions and 0 deletions

View File

@ -0,0 +1,12 @@
from graphene.contrib.sqlalchemy.types import (
SQLAlchemyObjectType,
SQLAlchemyInterface,
SQLAlchemyNode
)
from graphene.contrib.sqlalchemy.fields import (
SQLAlchemyConnectionField,
SQLAlchemyModelField
)
__all__ = ['SQLAlchemyObjectType', 'SQLAlchemyInterface', 'SQLAlchemyNode',
'SQLAlchemyConnectionField', 'SQLAlchemyModelField']

View File

@ -0,0 +1,61 @@
from sqlalchemy import types
from sqlalchemy.orm import interfaces
from singledispatch import singledispatch
from graphene.contrib.sqlalchemy.fields import ConnectionOrListField, SQLAlchemyModelField
from graphene.core.fields import BooleanField, FloatField, IDField, IntField, StringField
def convert_sqlalchemy_relationship(relationship):
model_field = SQLAlchemyModelField(field.table, description=relationship.key)
if relationship.direction == interfaces.ONETOMANY:
return model_field
elif (relationship.direction == interfaces.MANYTOONE or
relationship.direction == interfaces.MANYTOMANY):
return ConnectionOrListField(model_field)
def convert_sqlalchemy_column(column):
try:
return convert_sqlalchemy_type(column.type, column)
except Exception:
raise
@singledispatch
def convert_sqlalchemy_type():
raise Exception(
"Don't know how to convert the SQLAlchemy column %s (%s)" % (column, column.__class__))
@convert_sqlalchemy_type.register(types.Date)
@convert_sqlalchemy_type.register(types.DateTime)
@convert_sqlalchemy_type.register(types.Time)
@convert_sqlalchemy_type.register(types.Text)
@convert_sqlalchemy_type.register(types.String)
@convert_sqlalchemy_type.register(types.Unicode)
@convert_sqlalchemy_type.register(types.UnicodeText)
@convert_sqlalchemy_type.register(types.Enum)
def convert_column_to_string(type, column):
return StringField(description=column.description)
@convert_sqlalchemy_type.register(types.SmallInteger)
@convert_sqlalchemy_type.register(types.BigInteger)
@convert_sqlalchemy_type.register(types.Integer)
def convert_column_to_int_or_id(column):
if column.primary_key:
return IDField(description=column.description)
else:
return IntField(description=column.description)
@convert_sqlalchemy_type.register(types.Boolean)
def convert_column_to_boolean(column):
return BooleanField(description=column.description)
@convert_sqlalchemy_type.register(types.Float)
@convert_sqlalchemy_type.register(types.Numeric)
def convert_column_to_float(column):
return FloatField(description=column.description)

View File

@ -0,0 +1,67 @@
from graphene import relay
from graphene.contrib.sqlalchemy.utils import get_type_for_model, lazy_map
from graphene.core.fields import Field, LazyField, ListField
from graphene.relay.utils import is_node
class SQLAlchemyConnectionField(relay.ConnectionField):
def wrap_resolved(self, value, instance, args, info):
schema = info.schema.graphene_schema
return lazy_map(value, self.get_object_type(schema))
class LazyListField(ListField):
def resolve(self, instance, args, info):
schema = info.schema.graphene_schema
resolved = super(LazyListField, self).resolve(instance, args, info)
return lazy_map(resolved, self.get_object_type(schema))
class ConnectionOrListField(LazyField):
def get_field(self, schema):
model_field = self.field_type
field_object_type = model_field.get_object_type(schema)
if is_node(field_object_type):
field = SQLAlchemyConnectionField(model_field)
else:
field = LazyListField(model_field)
field.contribute_to_class(self.object_type, self.name)
return field
class SQLAlchemyModelField(Field):
def __init__(self, model, *args, **kwargs):
super(SQLAlchemyModelField, self).__init__(None, *args, **kwargs)
self.model = model
def resolve(self, instance, args, info):
resolved = super(SQLAlchemyModelField, self).resolve(instance, args, info)
schema = info.schema.graphene_schema
_type = self.get_object_type(schema)
assert _type, ("Field %s cannot be retrieved as the "
"ObjectType is not registered by the schema" % (
self.attname
))
return _type(resolved)
def internal_type(self, schema):
_type = self.get_object_type(schema)
if not _type and self.object_type._meta.only_fields:
raise Exception(
"Model %r is not accessible by the schema. "
"You can either register the type manually "
"using @schema.register. "
"Or disable the field %s in %s" % (
self.model,
self.attname,
self.object_type
)
)
return schema.T(_type) or Field.SKIP
def get_object_type(self, schema):
return get_type_for_model(schema, self.model)

View File

@ -0,0 +1,37 @@
import inspect
from sqlalchemy import Table
from graphene.core.options import Options
from graphene.relay.types import Node
from graphene.relay.utils import is_node
VALID_ATTRS = ('table', 'only_columns', 'exclude_columns')
def is_base(cls):
from graphene.contrib.SQLAlchemy.types import SQLAlchemyObjectType
return SQLAlchemyObjectType in cls.__bases__
class SQLAlchemyOptions(Options):
def __init__(self, *args, **kwargs):
self.table = None
super(SQLAlchemyOptions, self).__init__(*args, **kwargs)
self.valid_attrs += VALID_ATTRS
self.only_fields = None
self.exclude_fields = []
def contribute_to_class(self, cls, name):
super(SQLAlchemyOptions, self).contribute_to_class(cls, name)
if is_node(cls):
self.exclude_fields = list(self.exclude_fields)
self.interfaces.append(Node)
if not is_node(cls) and not is_base(cls):
return
if not self.table:
raise Exception(
'SQLAlchemy ObjectType %s must have a table in the Meta class attr' % cls)
elif not inspect.isclass(self.table) or not issubclass(self.table, Table):
raise Exception('Provided table in %s is not a SQLAlchemy table' % cls)

View File

@ -0,0 +1,56 @@
import six
from sqlalchemy.inspection import inspect
from graphene.contrib.sqlalchemy.converter import convert_sqlalchemy_column,
convert_sqlalchemy_relationship
from graphene.contrib.sqlalchemy.options import SQLAlchemyOptions
from graphene.contrib.sqlalchemy.utils import get_reverse_columns
from graphene.core.types import BaseObjectType, ObjectTypeMeta
from graphene.relay.fields import GlobalIDField
from graphene.relay.types import BaseNode
class SQLAlchemyObjectTypeMeta(ObjectTypeMeta):
options_cls = SQLAlchemyOptions
def is_interface(cls, parents):
return SQLAlchemyInterface in parents
def add_extra_fields(cls):
if not cls._meta.table:
return
inspected_table = inspect(cls._meta.table)
# Get all the columns for the relationships on the table
for relationship in inspected_table.relationships:
converted_relationship = convert_sqlalchemy_relationship(relationship)
cls.add_to_class(relationship.key, converted_relationship)
for column in inspected_table.columns:
converted_column = convert_sqlalchemy_column(column)
cls.add_to_class(column.name, converted_column)
class InstanceObjectType(BaseObjectType):
def __init__(self, instance=None):
self.instance = instance
super(InstanceObjectType, self).__init__()
def __getattr__(self, attr):
return getattr(self.instance, attr)
class SQLAlchemyObjectType(six.with_metaclass(SQLAlchemyObjectTypeMeta, InstanceObjectType)):
pass
class SQLAlchemyInterface(six.with_metaclass(SQLAlchemyObjectTypeMeta, InstanceObjectType)):
pass
class SQLAlchemyNode(BaseNode, SQLAlchemyInterface):
id = GlobalIDField()
@classmethod
def get_node(cls, id):
instance = cls._meta.table.objects.filter(id=id).first()
return cls(instance)

View File

@ -0,0 +1,23 @@
from django.db import models
from django.db.models.manager import Manager
from django.db.models.query import QuerySet
from graphene.utils import LazyMap
def get_type_for_model(schema, model):
schema = schema
types = schema.types.values()
for _type in types:
type_model = hasattr(_type, '_meta') and getattr(
_type._meta, 'model', None)
if model == type_model:
return _type
def lazy_map(value, func):
if isinstance(value, Manager):
value = value.get_queryset()
if isinstance(value, QuerySet):
return LazyMap(value, func)
return value

View File

@ -70,6 +70,9 @@ setup(
'singledispatch>=3.4.0.3',
'graphql-django-view>=1.0.0',
],
'sqlalchemy': [
'SQLAlchemy'
]
},
cmdclass={'test': PyTest},