shit.py

from collections import OrderedDict

import graphene
from graphene import relay
from graphene.types.utils import yank_fields_from_attrs
from graphene_sqlalchemy import SQLAlchemyConnectionField, SQLAlchemyObjectType
from graphene_sqlalchemy.converter import (convert_sqlalchemy_column,
                                           convert_sqlalchemy_composite,
                                           convert_sqlalchemy_hybrid_method)
from graphene_sqlalchemy.fields import default_connection_field_factory
from graphene_sqlalchemy.registry import get_global_registry
from sqlalchemy.ext.hybrid import hybrid_property

from . import models


class Connection(relay.Connection):
  class Meta:
    abstract = True

  total_count = graphene.Int()

  def resolve_total_count(self, info, **kw):
    return self.length


# https://github.com/Rafik-Belkadi/graphene-sqlalchemy/blob/master/graphene_sqlalchemy/types.py#L214-L278
def construct_fields_for_input(obj_type, model, registry, connection_field_factory, only_fields, exclude_fields):
  inspected_model = sqlalchemyinspect(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 exclude this field in exclude_fields
      continue
    converted_column = convert_sqlalchemy_column(column, registry)
    fields[name] = converted_column
  for name, composite in inspected_model.composites.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 exclude this field in exclude_fields
      continue
    converted_composite = convert_sqlalchemy_composite(composite, registry)
    fields[name] = converted_composite
  for hybrid_item in inspected_model.all_orm_descriptors:
    if type(hybrid_item) == hybrid_property:
      name = hybrid_item.__name__
      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 exclude this field in exclude_fields
        continue
      converted_hybrid_property = convert_sqlalchemy_hybrid_method(hybrid_item)
      fields[name] = converted_hybrid_property
  return fields


class SQLAlchemyInputObjectType(graphene.InputObjectType):
  @classmethod
  def __init_subclass_with_meta__(cls, model=None, registry=None, skip_registry=False,only_fields=(), exclude_fields=(), connection=None,_meta=None, connection_field_factory=default_connection_field_factory, use_connection=None, interfaces=(), id=None, **options):
    if not registry:
      registry = get_global_registry()
    sqla_fields = yank_fields_from_attrs(
      construct_fields_for_input(model=model, registry=registry,only_fields=only_fields, exclude_fields=exclude_fields,
              connection_field_factory=connection_field_factory,obj_type=cls),
      _as=graphene.Field,
    )


class Todo(SQLAlchemyObjectType):
  class Meta:
    model = models.Todo
    interfaces = (relay.Node, )


class TodoConnection(Connection):
  class Meta:
    node = Todo


class CreateTodoInput(SQLAlchemyInputObjectType):
  class Model:
    model = models.Todo
    exclude_fields = ('id', )


class CreateTodo(graphene.Mutation):
  class Arguments:
    input = CreateTodoInput(required=True)

  @classmethod
  def mutate(cls, instance, info, **args):



class Query(graphene.ObjectType):
  node = relay.Node.Field()
  todos = SQLAlchemyConnectionField(TodoConnection)


schema = graphene.Schema(query=Query)

mutation {
  deleteContact(
    id: "Q29udGFjdE5vZGU6Mw=="
  ) {
    ok
  }
}