Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_engines_cache(self):
from aldjemy.core import Cache, get_engine
self.assertEqual(get_engine('default'), Cache.engines['default'])
self.assertEqual(get_engine('logs'), Cache.engines['logs'])
self.assertEqual(get_engine(), Cache.engines['default'])
self.assertNotEqual(get_engine('default'), get_engine('logs'))
def test_engines_cache(self):
from aldjemy.core import Cache, get_engine
self.assertEqual(get_engine('default'), Cache.engines['default'])
self.assertEqual(get_engine('logs'), Cache.engines['logs'])
self.assertEqual(get_engine(), Cache.engines['default'])
self.assertNotEqual(get_engine('default'), get_engine('logs'))
def get_engine(alias='default'):
if alias not in Cache.engines:
engine_string = get_engine_string(alias)
# we have to use autocommit=True, because SQLAlchemy
# is not aware of Django transactions
kw = {}
if engine_string == 'sqlite3':
kw['native_datetime'] = True
pool = DjangoPool(alias=alias, creator=None)
Cache.engines[alias] = create_engine(get_connection_string(alias),
pool=pool, **kw)
return Cache.engines[alias]
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,
def get_meta():
if not getattr(Cache, 'meta', None):
Cache.meta = MetaData()
return Cache.meta