graphene/graphene-sqlalchemy/graphene_sqlalchemy/converter.py

134 lines
4.8 KiB
Python
Raw Normal View History

2016-07-18 18:28:12 +03:00
from singledispatch import singledispatch
from sqlalchemy import types
from sqlalchemy.orm import interfaces
from sqlalchemy.dialects import postgresql
from graphene import Enum, ID, Boolean, Float, Int, String, List, Field, Dynamic
from graphene.relay import is_node
2016-07-18 18:28:12 +03:00
from graphene.types.json import JSONString
from .fields import SQLAlchemyConnectionField
2016-07-18 18:28:12 +03:00
try:
from sqlalchemy_utils import ChoiceType, ScalarListType
2016-07-18 18:28:12 +03:00
except ImportError:
class ChoiceType(object):
pass
class ScalarListType(object):
pass
2016-07-18 18:28:12 +03:00
def convert_sqlalchemy_relationship(relationship, registry):
2016-07-18 18:28:12 +03:00
direction = relationship.direction
model = relationship.mapper.entity
def dynamic_type():
_type = registry.get_type_for_model(model)
if not _type:
return None
if (direction == interfaces.MANYTOONE or not relationship.uselist):
return Field(_type)
elif (direction == interfaces.ONETOMANY or
direction == interfaces.MANYTOMANY):
if is_node(_type):
return SQLAlchemyConnectionField(_type)
return Field(List(_type))
return Dynamic(dynamic_type)
2016-07-18 18:28:12 +03:00
2016-09-01 00:39:16 +03:00
def convert_sqlalchemy_composite(composite, registry):
converter = registry.get_converter_for_composite(composite.composite_class)
if not converter:
try:
raise Exception(
"Don't know how to convert the composite field %s (%s)" %
(composite, composite.composite_class))
except AttributeError:
# handle fields that are not attached to a class yet (don't have a parent)
raise Exception(
"Don't know how to convert the composite field %r (%s)" %
(composite, composite.composite_class))
return converter(composite, registry)
def _register_composite_class(cls, registry=None):
if registry is None:
from .registry import get_global_registry
registry = get_global_registry()
def inner(fn):
registry.register_composite_converter(cls, fn)
return inner
convert_sqlalchemy_composite.register = _register_composite_class
def convert_sqlalchemy_column(column, registry=None):
return convert_sqlalchemy_type(getattr(column, 'type', None), column, registry)
2016-07-18 18:28:12 +03:00
@singledispatch
def convert_sqlalchemy_type(type, column, registry=None):
2016-07-18 18:28:12 +03:00
raise Exception(
"Don't know how to convert the SQLAlchemy field %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.String)
@convert_sqlalchemy_type.register(types.Text)
@convert_sqlalchemy_type.register(types.Unicode)
@convert_sqlalchemy_type.register(types.UnicodeText)
@convert_sqlalchemy_type.register(types.Enum)
@convert_sqlalchemy_type.register(postgresql.ENUM)
@convert_sqlalchemy_type.register(postgresql.UUID)
def convert_column_to_string(type, column, registry=None):
2016-08-16 09:46:30 +03:00
return String(description=column.doc, required=not(column.nullable))
2016-07-18 18:28:12 +03:00
@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(type, column, registry=None):
2016-07-18 18:28:12 +03:00
if column.primary_key:
2016-08-16 09:46:30 +03:00
return ID(description=column.doc, required=not(column.nullable))
2016-07-18 18:28:12 +03:00
else:
2016-08-16 09:46:30 +03:00
return Int(description=column.doc, required=not(column.nullable))
2016-07-18 18:28:12 +03:00
@convert_sqlalchemy_type.register(types.Boolean)
def convert_column_to_boolean(type, column, registry=None):
2016-08-16 09:46:30 +03:00
return Boolean(description=column.doc, required=not(column.nullable))
2016-07-18 18:28:12 +03:00
@convert_sqlalchemy_type.register(types.Float)
@convert_sqlalchemy_type.register(types.Numeric)
def convert_column_to_float(type, column, registry=None):
2016-08-16 09:46:30 +03:00
return Float(description=column.doc, required=not(column.nullable))
2016-07-18 18:28:12 +03:00
@convert_sqlalchemy_type.register(ChoiceType)
def convert_column_to_enum(type, column, registry=None):
2016-07-18 18:28:12 +03:00
name = '{}_{}'.format(column.table.name, column.name).upper()
return Enum(name, type.choices, description=column.doc)
@convert_sqlalchemy_type.register(ScalarListType)
def convert_scalar_list_to_list(type, column, registry=None):
return List(String, description=column.doc)
2016-07-18 18:28:12 +03:00
@convert_sqlalchemy_type.register(postgresql.ARRAY)
def convert_postgres_array_to_list(type, column, registry=None):
2016-07-18 18:28:12 +03:00
graphene_type = convert_sqlalchemy_type(column.type.item_type, column)
2016-08-16 09:46:30 +03:00
return List(graphene_type, description=column.doc, required=not(column.nullable))
2016-07-18 18:28:12 +03:00
@convert_sqlalchemy_type.register(postgresql.HSTORE)
@convert_sqlalchemy_type.register(postgresql.JSON)
@convert_sqlalchemy_type.register(postgresql.JSONB)
def convert_json_to_string(type, column, registry=None):
2016-08-16 09:46:30 +03:00
return JSONString(description=column.doc, required=not(column.nullable))