sqlalchemy_reflect_tables_to_types.py

import json

from sqlalchemy import inspect as sqla_inspect
from sqlalchemy.orm.dependency import OneToManyDP, ManyToOneDP
from sqlalchemy.exc import NoInspectionAvailable
from sqlalchemy.orm.mapper import Mapper
from sqlalchemy import Column, String, Integer, create_engine, ForeignKey, inspect
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship


engine = create_engine('sqlite:///:memory:', echo=False)
Session = sessionmaker()
Session.configure(bind=engine)
session = Session()
Base = declarative_base()


class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String)

    teams = relationship('Team', secondary='team_member')
    member_of = relationship('TeamMember')

    team1 = relationship('Team', secondary='team_member')
    team2 = relationship('Team', secondary='team_member')


class Team(Base):
    __tablename__ = 'team'
    id = Column(Integer, primary_key=True)
    name = Column(String)

    members = relationship('TeamMember')


class TeamMember(Base):
    __tablename__ = 'team_member'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    user_id = Column(Integer, ForeignKey('user.id'), nullable=False)
    team_id = Column(Integer, ForeignKey('team.id'), nullable=False)

    team = relationship('Team')
    result = relationship('User')

Base.metadata.create_all(engine)


def reflect_sqla(models):
    """
    SQLAlchemy models to JSON. Argument `models` should
    either be a module:

      import my_app.models as models
      reflect_sqla(models)

    Or a list of strings, in this case
    a listing of attributes of the current
    module:

      reflect_sqla(dir())
    """
    from inspect import ismodule

    classes = {}
    if ismodule(models):
        _models = [getattr(models, attr) for attr in dir(models)]
    else:
        _models = [globals()[attr] for attr in models]

    models = []
    for _model in _models:
        try:
            model = sqla_inspect(_model)
            if isinstance(model, Mapper):
                models.append(model)
        except NoInspectionAvailable:
            continue

    for model in models:
        data = {}
        for column in model.columns:
            try:
                data[column.name] = column.type.python_type.__name__
            except NotImplementedError:
                continue

        if model.relationships:
            for column, relationship in dict(model.relationships).items():
                dep = relationship._dependency_processor
                if isinstance(dep, OneToManyDP):
                    data[column] = list.__name__
                elif isinstance(dep, ManyToOneDP):
                    data[column] = next(iter(relationship.remote_side)).type.python_type.__name__

        classes[model.mapped_table.name] = data

    return classes

result = reflect_sqla(dir())
print(json.dumps(result, indent=4, sort_keys=True))

# {
#     "team": {
#         "id": "int",
#         "members": "list",
#         "name": "str"
#     },
#     "team_member": {
#         "id": "int",
#         "name": "str",
#         "result": "int",
#         "team": "int",
#         "team_id": "int",
#         "user_id": "int"
#     },
#     "user": {
#         "id": "int",
#         "member_of": "list",
#         "name": "str"
#     }
# }