Dynamic generation of graphql interface based on polymorphic ORM inheritance

gql_poly.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Boolean, String, String, Integer, ForeignKey
from sqlalchemy.orm import sessionmaker
from secrets import token_hex
from json import dumps
import graphene
from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField
from graphene.relay import Connection
from collections import defaultdict

engine = create_engine('sqlite:///:memory:', echo=True)
session = sessionmaker(bind=engine)()
Base = declarative_base()

class UserModel(Base):
  __tablename__ = 'user'
  _id     = Column(Integer, primary_key=True)
  role    = Column(String, nullable=False)
  name    = Column(String)
  enabled = Column(Boolean)

  __mapper_args__ = {
    'polymorphic_identity': 'user',
    'polymorphic_on': 'role'
  }

  def __repr__(self):
    return f'{self.role.capitalize()}({self.name}, enabled={self.enabled})'

class AdminModel(UserModel):
  __tablename__   = 'admin'
  __mapper_args__ = { 'polymorphic_identity': 'admin' }
  _id = Column(Integer, ForeignKey('user._id'), primary_key=True)
  admin_secret = Column(String(30), default=token_hex(6))

class SuperAdminModel(AdminModel):
  __tablename__   = 'superadmin'
  __mapper_args__ = { 'polymorphic_identity': 'superadmin' }
  _id = Column(Integer, ForeignKey('admin._id'), primary_key=True)
  super_secret = Column(String(30), default=token_hex(6))

Base.metadata.create_all(engine)
session.add(AdminModel(name='a', role='admin', enabled=False))
session.add(UserModel(name='b', role='user', enabled=False))
session.add(UserModel(name='c', role='user', enabled=True))
session.add(SuperAdminModel(name='d', role='superadmin', enabled=False))
session.commit()

# A query on User will update the Admin as well.
enabled = ['a', 'd']
session.query(UserModel).update({UserModel.enabled:UserModel.name.in_(enabled)}, synchronize_session=False)

for user in session.query(UserModel):
  print(f'{user}')

def get_class_map(classes):
  class_map = dict()
  classes = list(classes)
  while classes:
    cls = classes.pop()
    subclasses = class_map[cls] = list()
    _get_class_map(cls, subclasses, classes)
  return class_map

def _get_class_map(cls, subclasses, allclasses):
  for sub in cls.__subclasses__():
    subclasses.append(sub)
    allclasses.append(sub)
    _get_class_map(sub, subclasses, allclasses)

def make_resolve_func(cls):
  def resolve_func(self, info):
    query = cls.get_query(info)
    return query.all()
  return resolve_func

def make_schema():

  class_map = get_class_map(Base.__subclasses__())

  queries = dict()
  for cls, subs in class_map.items():
    name       = cls.__name__.replace('Model', '')
    conn_name  = f'{name}Connection'
    query_name = f'{name.lower()}s'

    SqlType  = type(name, (SQLAlchemyObjectType,), {
      'Meta': type('Meta',(), {'model': cls})})
    SqlTypeConnection = type(conn_name, (Connection,), {
      'Meta': type('Meta',(), {'node': SqlType})})

    print(f'{name} => {query_name} => {cls.__name__}')
    queries[f'relay_{query_name}'] = SQLAlchemyConnectionField(SqlTypeConnection)
    queries[query_name] = graphene.List(SqlType)
    queries[f'resolve_{query_name}'] = make_resolve_func(SqlType)

  Query = type('Query', (graphene.ObjectType, ), queries)
  schema = graphene.Schema(query=Query)
  return schema

schema = make_schema()
with open('schema.gql', 'w') as fd:
  fd.write(str(schema))

query = '''
query {
  users {
    name,
    role,
    enabled
  }
}
'''
result = schema.execute(query, context_value={'session': session})
print(dumps(result.to_dict()))

query = '''
query {
  relayUsers {
    edges {
      node {
        name,
        role,
        enabled
      }
    }
  }
}
'''
result = schema.execute(query, context_value={'session': session})
print(dumps(result.to_dict()))

schema.gql

schema {
  query: Query
}

type Admin {
  Id: ID!
  role: String!
  name: String
  enabled: Boolean
  adminSecret: String
}

type AdminConnection {
  pageInfo: PageInfo!
  edges: [AdminEdge]!
}

type AdminEdge {
  node: Admin
  cursor: String!
}

enum AdminSortEnum {
  _ID_ASC
  _ID_DESC
  ROLE_ASC
  ROLE_DESC
  NAME_ASC
  NAME_DESC
  ENABLED_ASC
  ENABLED_DESC
  ADMIN_SECRET_ASC
  ADMIN_SECRET_DESC
}

type PageInfo {
  hasNextPage: Boolean!
  hasPreviousPage: Boolean!
  startCursor: String
  endCursor: String
}

type Query {
  relayUsers(sort: [UserSortEnum] = [_ID_ASC], before: String, after: String, first: Int, last: Int): UserConnection
  users: [User]
  relaySuperadmins(sort: [SuperAdminSortEnum] = [_ID_ASC], before: String, after: String, first: Int, last: Int): SuperAdminConnection
  superadmins: [SuperAdmin]
  relayAdmins(sort: [AdminSortEnum] = [_ID_ASC], before: String, after: String, first: Int, last: Int): AdminConnection
  admins: [Admin]
}

type SuperAdmin {
  Id: ID!
  role: String!
  name: String
  enabled: Boolean
  adminSecret: String
  superSecret: String
}

type SuperAdminConnection {
  pageInfo: PageInfo!
  edges: [SuperAdminEdge]!
}

type SuperAdminEdge {
  node: SuperAdmin
  cursor: String!
}

enum SuperAdminSortEnum {
  _ID_ASC
  _ID_DESC
  ROLE_ASC
  ROLE_DESC
  NAME_ASC
  NAME_DESC
  ENABLED_ASC
  ENABLED_DESC
  ADMIN_SECRET_ASC
  ADMIN_SECRET_DESC
  SUPER_SECRET_ASC
  SUPER_SECRET_DESC
}

type User {
  Id: ID!
  role: String!
  name: String
  enabled: Boolean
}

type UserConnection {
  pageInfo: PageInfo!
  edges: [UserEdge]!
}

type UserEdge {
  node: User
  cursor: String!
}

enum UserSortEnum {
  _ID_ASC
  _ID_DESC
  ROLE_ASC
  ROLE_DESC
  NAME_ASC
  NAME_DESC
  ENABLED_ASC
  ENABLED_DESC
}