-
Notifications
You must be signed in to change notification settings - Fork 170
/
utils.py
176 lines (150 loc) · 7.21 KB
/
utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
"""Various utility functions for registering and activating models."""
import webbrowser
from flask import current_app
from flask.ext.admin import Admin
from flask.ext.admin.contrib.sqla import ModelView
from sqlalchemy.engine import reflection
from sqlalchemy.ext.declarative import declarative_base, DeferredReflection
from sqlalchemy.orm import relationship
from sqlalchemy.schema import Table
from sandman import app, db
from sandman.model.models import Model, AdminModelViewWithPK
def generate_endpoint_classes(db, generate_pks=False):
"""Return a list of model classes generated for each reflected database
table."""
seen_classes = set()
with app.app_context():
db.metadata.reflect(bind=db.engine)
for name, table in db.metadata.tables.items():
if not name in seen_classes:
seen_classes.add(name)
if not table.primary_key and generate_pks:
cls = add_pk_if_required(db, table, name)
else:
cls = type(str(name), (sandman_model, db.Model),
{'__tablename__': name})
register(cls)
def add_pks_if_required(db, known_tables=None):
for name, table in db.metadata.tables.items():
if known_tables is None or table in known_tables:
cls = add_pk_if_required(db, table, name)
register(cls)
def add_pk_if_required(db, table, name):
"""Return a class deriving from our Model class as well as the SQLAlchemy
model.
:param `sqlalchemy.schema.Table` table: table to create primary key for
:param table: table to create primary key for
"""
db.metadata.reflect(bind=db.engine)
cls_dict = {'__tablename__': name}
if not table.primary_key:
for column in table.columns:
column.primary_key = True
Table(name, db.metadata, *table.columns, extend_existing=True)
cls_dict['__table__'] = table
db.metadata.create_all(bind=db.engine)
return type(str(name), (sandman_model, db.Model), cls_dict)
def prepare_relationships(db, known_tables):
"""Enrich the registered Models with SQLAlchemy ``relationships``
so that related tables are correctly processed up by the admin.
"""
inspector = reflection.Inspector.from_engine(db.engine)
for cls in set(known_tables.values()):
for foreign_key in inspector.get_foreign_keys(cls.__tablename__):
other = known_tables[foreign_key['referred_table']]
if other not in cls.__related_tables__ and cls not in other.__related_tables__ and other != cls:
cls.__related_tables__.add(other)
# Add a SQLAlchemy relationship as an attribute on the class
setattr(cls, other.__name__.lower(), relationship(
other.__name__, backref=cls.__name__.lower()))
def register(cls, use_admin=True):
"""Register with the API a :class:`sandman.model.Model` class and associated
endpoint.
:param cls: User-defined class derived from :class:`sandman.model.Model` to
be registered with the endpoint returned by :func:`endpoint()`
:type cls: :class:`sandman.model.Model` or tuple
"""
with app.app_context():
if getattr(current_app, 'class_references', None) is None:
current_app.class_references = {}
if isinstance(cls, (list, tuple)):
for entry in cls:
register_internal_data(entry)
entry.use_admin = use_admin
else:
register_internal_data(cls)
cls.use_admin = use_admin
def register_internal_data(cls):
"""Register a new class, *cls*, with various internal data structures.
:params `sandman.model.Model` cls: class to register
"""
with app.app_context():
current_app.class_references[cls.__tablename__] = cls
current_app.class_references[cls.__name__] = cls
current_app.class_references[cls.endpoint()] = cls
if not getattr(cls, '__related_tables__', None):
cls.__related_tables__ = set()
def register_classes_for_admin(db_session, show_pks=True,
name='admin'):
"""Registers classes for the Admin view that ultimately creates the admin
interface.
:param db_session: handle to database session
:param list classes: list of classes to register with the admin
:param bool show_pks: show primary key columns in the admin?
"""
with app.app_context():
admin_view = Admin(current_app, name=name)
for cls in set(cls for cls in current_app.class_references.values() if
cls.use_admin == True):
column_list = [column.name for column in
cls.__table__.columns.values()]
if hasattr(cls, '__view__'):
# allow ability for model classes to specify model views
admin_view_class = type('AdminView', (cls.__view__,),
{'form_columns': column_list})
elif show_pks:
# the default of Flask-SQLAlchemy is to not show primary
# classes, which obviously isn't acceptable in some cases
admin_view_class = type('AdminView', (AdminModelViewWithPK,),
{'form_columns': column_list})
else:
admin_view_class = ModelView
admin_view.add_view(admin_view_class(cls, db_session))
def activate(admin=True, browser=True, name='admin'):
"""Activate each pre-registered model or generate the model classes and
(possibly) register them for the admin.
:param bool admin: should we generate the admin interface?
:param bool browser: should we open the browser for the user?
:param name: name to use for blueprint created by the admin interface. Set
this to avoid naming conflicts with other blueprints (if trying
to use sandman to connect to multiple databases simultaneously)
"""
with app.app_context():
if getattr(current_app, 'class_references', None) is None:
current_app.class_references = {}
try:
generate_pks = current_app.config['SANDMAN_GENERATE_PKS']
except KeyError:
generate_pks = False
generate_endpoint_classes(db, generate_pks)
else:
Model.prepare(db.engine)
prepare_relationships(db, current_app.class_references)
if admin:
try:
show_pks = current_app.config['SANDMAN_SHOW_PKS']
except KeyError:
show_pks = False
register_classes_for_admin(db.session, show_pks, name)
print current_app.url_map
if browser:
webbrowser.open('http://127.0.0.1:5000/admin')
# Redefine 'Model' to be a sqlalchemy.ext.declarative.api.DeclarativeMeta
# object which also derives from sandman.models.Model. The naming is done for
# documentation/clarity purposes. Previously the line below had Model deriving
# from DeferredReflection and "Resource", which was the exact same class
# as "Model" in models.py. It caused confusion in the documentation, however,
# since it wasn't clear that the Model class and the Resource class were
# actually the same thing.
sandman_model = Model
Model = declarative_base(cls=(Model, DeferredReflection))