Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def construct_models(metadata):
if not metadata.tables:
generate_tables(metadata)
tables = metadata.tables
models = [model for model in get_all_django_models() if not model._meta.proxy]
sa_models_by_django_models = {}
for model in models:
table_name = (
metadata.schema + '.' + model._meta.db_table
if metadata.schema else model._meta.db_table
)
mixin = getattr(model, 'aldjemy_mixin', None)
bases = (mixin, BaseSQLAModel) if mixin else (BaseSQLAModel, )
table = tables[table_name]
# because querying happens on sqlalchemy side, we can use only one
# type of queries for alias, so we use 'read' type
sa_model = type(model._meta.object_name, bases,
{'table': table,
'alias': router.db_for_read(model)})
sa_models_by_django_models[model] = sa_model
for model in models:
sa_model = sa_models_by_django_models[model]
table_name = (
metadata.schema + '.' + model._meta.db_table
if metadata.schema else model._meta.db_table
)
def prepare_models():
tables = get_tables()
models = get_django_models()
sa_models_by_django_models = getattr(Cache, 'sa_models', {})
with warnings.catch_warnings():
warnings.simplefilter("ignore")
sa_models_by_table_names = getattr(Cache, 'models', {})
for model in models:
table_name = model._meta.db_table
mixin = getattr(model, 'aldjemy_mixin', None)
bases = (mixin, BaseSQLAModel) if mixin else (BaseSQLAModel, )
table = tables[table_name]
# because querying happens on sqlalchemy side, we can use only one
# type of queries for alias, so we use 'read' type
sa_model = type(model._meta.object_name, bases,
{'table': table,
'alias': router.db_for_read(model)})
sa_models_by_table_names[table_name] = sa_model
sa_models_by_django_models[model] = sa_model
for model in models:
sa_model = sa_models_by_django_models[model]
table = tables[model._meta.db_table]
attrs = _extract_model_attrs(model, sa_models_by_django_models)
orm.mapper(sa_model, table, attrs)