"""Database level utility class. Used for constructing and seeding the test SQLite database. """ from datetime import datetime import sys import sqlalchemy from social_analytics import config from social_analytics.connectors import mysql from social_analytics.models import social_profile from social_analytics.models import artist_social_profile from social_analytics.models import artist_url # TODO fix tests for this query query_social_profiles_to_collect = ( '''SELECT social_profile_id, platform, platform_id, platform_name, collection_scheduled_time, last_collected_date, created_date, updated_date, created_by, updated_by FROM social_profile sp WHERE sp.collection_scheduled_time >= TIME(DATE_SUB(NOW(), INTERVAL 1 HOUR)) AND sp.last_collected <= NOW() ORDER BY sp.collection_scheduled_time ASC LIMIT 500;''' ) social_profile_data = [ { 'social_profile_id': 1, 'platform': 'facebook', 'platform_id': 123, 'platform_name': 'Anika Pyle', 'last_collected_date': datetime(2017, 11, 16, 9, 38) }, { 'social_profile_id': 2, 'platform': 'instagram', 'platform_id': 345, 'platform_name': 'anika_pyle', 'last_collected_date': datetime(2017, 11, 16, 9, 38) }, { 'social_profile_id': 3, 'platform': 'facebook', 'platform_id': 456, 'platform_name': 'Anika Pyle', 'last_collected_date': datetime(2017, 11, 16, 10, 38) }, { 'social_profile_id': 4, 'platform': 'facebook', 'platform_id': 678, 'platform_name': 'Taylor Swift', 'last_collected_date': datetime(2017, 11, 16, 10, 38) }, { 'social_profile_id': 6, 'platform': 'facebook', 'platform_id': 912, 'platform_name': 'Test Artist', 'last_collected_date': None }, { 'social_profile_id': 7, 'platform': 'facebook', 'platform_id': 103, 'platform_name': 'Test Artist 2', 'last_collected_date': datetime(2017, 11, 15, 10, 38) }, { 'social_profile_id': 19, 'platform': 'instagram', 'platform_id': 184, 'platform_name': 'Test Artist 3', 'last_collected_date': datetime(2017, 11, 15, 10, 38) } ] artist_social_profile_data = [ { 'artist_id': 1, 'social_profile_id': 1, 'label_id': 1 }, { 'artist_id': 2, 'social_profile_id': 2, 'label_id': 2 }, { 'artist_id': 3, 'social_profile_id': 4, 'label_id': 2 }, { 'artist_id': 4, 'social_profile_id': 3, 'label_id': 3 }, { 'artist_id': 5, 'social_profile_id': 3, 'label_id': 7 } ] artist_url_data = [ { 'url_id': 1, 'artist_id': 1, 'social_profile_id': 1, 'site_id': 2, 'url': 'www.facebook.com/123' }, { 'url_id': 2, 'artist_id': 1, 'social_profile_id': 11, 'site_id': 3, 'url': 'www.instagram.com/123' }, { 'url_id': 3, 'artist_id': 2, 'social_profile_id': 5, 'site_id': 3, 'url': 'www.instagram.com/123' }, { 'url_id': 4, 'artist_id': 3, 'social_profile_id': 4, 'site_id': 3, 'url': 'www.instagram.com/123' }, { 'url_id': 5, 'artist_id': 289, 'social_profile_id': 6, 'site_id': 3, 'url': 'www.instagram.com/123' }, { 'url_id': 6, 'artist_id': 577, 'social_profile_id': 7, 'site_id': 3, 'url': 'www.instagram.com/123' }, { 'url_id': 7, 'artist_id': 285, 'social_profile_id': 18, 'site_id': 3, 'url': 'www.instagram.com/123' }, { 'url_id': 8, 'artist_id': 288, 'social_profile_id': 19, 'site_id': 3, 'url': 'www.instagram.com/123' }, { 'url_id': 9, 'artist_id': 293, 'social_profile_id': 20, 'site_id': 3, 'url': 'www.instagram.com/123' } ] def create_tables(): """Create tables defined in social_analytics.models.""" if config.ENVIRONMENT != config.TEST_ENVIRONMENT: return if config.DB_URL != 'sqlite://': return mysql.BaseModel.metadata.drop_all(mysql._db_engine) mysql.BaseModel.metadata.create_all(mysql._db_engine) @mysql.autosession() def seed_data(session): """Insert test data into social_profile table.""" if config.ENVIRONMENT != config.TEST_ENVIRONMENT: return if config.DB_URL != 'sqlite://': return try: session.execute(social_profile.SocialProfile.__table__.insert().values( social_profile_data)) session.execute(artist_social_profile.ArtistSocialProfile.__table__ .insert().values(artist_social_profile_data)) session.execute(artist_url.ArtistUrl.__table__ .insert().values(artist_url_data)) except sqlalchemy.exc.CompileError as error: print('Seeding error: SQL compile:', error) except sqlalchemy.exc.IntegrityError as error: print('Seeding error: Integrity error:', error) except: print('Unexpected error while seeding:', sys.exc_info()[0]) session.commit()