"""User Model. This model represents a User """ import sqlalchemy from sqlalchemy.orm import relationship from sqlalchemy.sql import func from podcast.connectors import mysql from podcast.constants import error from podcast.constants.feature_flag import FEATURE_PODCAST_IA_RESTRUCTURE from podcast.constants.user import ADMIN, CHART_TYPE, ORGANIZATION, ROLES from podcast.models import network as network_model from podcast.models import podcast as podcast_model from podcast.models import show_family as show_family_model from podcast.models import user_podcast_settings as user_podcast_settings_model from podcast.utils import api_utils from podcast.utils import exc from podcast.utils import feature_flag_utils user_network_table = sqlalchemy.Table( 'user_network', mysql.BaseModel.metadata, sqlalchemy.Column('user_id', sqlalchemy.Integer, sqlalchemy.ForeignKey('user.id')), sqlalchemy.Column('network_id', sqlalchemy.Integer, sqlalchemy.ForeignKey('network.id')), ) user_podcast_table = sqlalchemy.Table( 'user_podcast', mysql.BaseModel.metadata, sqlalchemy.Column('user_id', sqlalchemy.Integer, sqlalchemy.ForeignKey('user.id')), sqlalchemy.Column('podcast_id', sqlalchemy.Integer, sqlalchemy.ForeignKey('podcast.id')), ) user_podcast_favorite_table = sqlalchemy.Table( 'user_podcast_favorite', mysql.BaseModel.metadata, sqlalchemy.Column('user_id', sqlalchemy.Integer, sqlalchemy.ForeignKey('user.id')), sqlalchemy.Column('podcast_id', sqlalchemy.Integer, sqlalchemy.ForeignKey('podcast.id')), ) user_show_family_table = sqlalchemy.Table( 'user_show_family', mysql.BaseModel.metadata, sqlalchemy.Column('user_id', sqlalchemy.Integer, sqlalchemy.ForeignKey('user.id')), sqlalchemy.Column('show_family_id', sqlalchemy.Integer, sqlalchemy.ForeignKey('show_family.id')), ) class UserChartFavorite(mysql.BaseModel): """Users favorite charts.""" __tablename__ = 'user_chart_favorite' user_id = sqlalchemy.Column(sqlalchemy.INTEGER, sqlalchemy.ForeignKey('user.id'), primary_key=True) store = sqlalchemy.Column(sqlalchemy.VARCHAR(255), nullable=False, primary_key=True) category = sqlalchemy.Column(sqlalchemy.VARCHAR(255), nullable=False, primary_key=True) country = sqlalchemy.Column(sqlalchemy.VARCHAR(255), nullable=False, primary_key=True) chart_type = sqlalchemy.Column(sqlalchemy.Enum(*CHART_TYPE), nullable=False, primary_key=True) class User(mysql.BaseModel): """User model.""" __tablename__ = 'user' id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True, autoincrement=True) # noqa uuid = sqlalchemy.Column(sqlalchemy.VARCHAR(200)) email = sqlalchemy.Column(sqlalchemy.VARCHAR(255), nullable=False) name = sqlalchemy.Column(sqlalchemy.VARCHAR(255)) active = sqlalchemy.Column(sqlalchemy.Boolean, nullable=False, default=True) all_networks = sqlalchemy.Column(sqlalchemy.Boolean, nullable=False, default=False) organization = sqlalchemy.Column(sqlalchemy.Enum(*ORGANIZATION), nullable=False) role = sqlalchemy.Column(sqlalchemy.Enum(*ROLES), nullable=True, default=ROLES[0]) language = sqlalchemy.Column(sqlalchemy.VARCHAR(2), nullable=False, default='en') profile_id = sqlalchemy.Column(sqlalchemy.Integer) networks = relationship('Network', secondary=user_network_table, lazy='joined') podcasts = relationship('Podcast', secondary=user_podcast_table, lazy='select') show_families = relationship(show_family_model.ShowFamily, secondary=user_show_family_table, lazy='select') favorite_podcasts = relationship('Podcast', secondary=user_podcast_favorite_table, lazy='joined') settings = relationship(user_podcast_settings_model.UserPodcastSettings, lazy='select', uselist=False) created_by = sqlalchemy.Column(sqlalchemy.VARCHAR(200)) updated_by = sqlalchemy.Column(sqlalchemy.VARCHAR(200)) created_date = sqlalchemy.Column(sqlalchemy.DateTime, default=func.now()) updated_date = sqlalchemy.Column(sqlalchemy.DateTime, default=func.now(), onupdate=func.now()) last_login = sqlalchemy.Column(sqlalchemy.DateTime) def to_dict(self, is_only_users=False): """Return the object as dictionary.""" base_dict = dict( id=self.id, uuid=self.uuid, email=self.email, name=self.name, active=self.active, language=self.language, role=self.role, organization=self.organization, profile_id=self.profile_id, networks=None, podcasts=None, show_families=None, created_by=self.created_by, updated_by=self.updated_by, created_date=self.created_date, updated_date=self.updated_date, all_networks=self.all_networks, last_login=self.last_login ) if not is_only_users: base_dict['networks'] = [network.to_dict() for network in self.networks] base_dict['podcasts'] = [podcast.to_dict() for podcast in self.podcasts] base_dict['show_families'] = [show_family.to_dict() for show_family in self.show_families] return base_dict def create_user(payload): """Create user object. Args: payload (dict): user data """ with mysql.pod_db_session() as session: existing_user = _get_user_by_uuid(payload['uuid'], session) if existing_user: raise exc.OwsError.bad_request(error.ERROR_MESSAGE_USER_ALREADY_EXISTS) network_ids = payload.pop('network_ids', None) podcast_ids = payload.pop('podcast_ids', None) show_family_ids = payload.pop('show_family_ids', None) current_user_id = api_utils.get_user_id() payload['created_by'] = current_user_id payload['updated_by'] = current_user_id new_user = User(**payload) if network_ids: new_user.networks = session.query(network_model.Network).filter( network_model.Network.id.in_(network_ids) ).all() if podcast_ids: new_user.podcasts = session.query(podcast_model.Podcast).filter( podcast_model.Podcast.id.in_(podcast_ids) ).all() if show_family_ids: new_user.show_families = session.query(show_family_model.ShowFamily).filter( show_family_model.ShowFamily.id.in_(show_family_ids) ).all() session.add(new_user) session.flush() user_podcast_settings_model.create_user_settings(new_user.id) return new_user.to_dict() def get_user_by_id(user_id, raw=False): """Get single user object. Args: user_id (str): user_id unique identifier """ with mysql.pod_db_session(read_only=True) as session: user = session.query(User).filter(User.id == user_id).first() if not user or not user.active: raise exc.OwsError.not_found('{} {}'.format(user_id, error.ERROR_MESSAGE_USER_NOT_FOUND)) if raw: return user return user.to_dict() def get_user_by_uuid(user_id): """Get single user object. Beginning of switch to UUID. Args: user_id (str): user_id unique identifier """ with mysql.pod_db_session(read_only=True) as session: user = _get_user_by_uuid(user_id, session) if not user: raise exc.OwsError.not_found('{} {}'.format(user_id, error.ERROR_MESSAGE_USER_NOT_FOUND)) return user.to_dict() def get_users(limit=0, offset=0, network_ids=None, podcast_ids=None, organization=None, exclude_admins=False): """Return all the users. Args: limit (int): how many podcasts to retrieve. offset (int): the offset (for pagination). Returns: dict: containing the paginated users. """ with mysql.pod_db_session(read_only=True) as session: users, total_records = _get_users( session, limit, offset, network_ids, podcast_ids, organization, exclude_admins) return { 'items': users, 'pagination': { 'total_records': total_records } } def _get_users(session, limit, offset, network_ids, podcast_ids, organization, exclude_admins): query = session.query(User).filter(User.active.is_(True)).filter(User.profile_id > 0) filter_by_network_ids = network_ids if network_ids is not None else [] filter_by_podcast_ids = podcast_ids if podcast_ids is not None else [] if len(filter_by_network_ids) or len(filter_by_podcast_ids): query = query.filter(sqlalchemy.or_( User.networks.any(network_model.Network.id.in_(filter_by_network_ids)), User.all_networks.is_(True), User.podcasts.any(podcast_model.Podcast.id.in_(filter_by_podcast_ids)), )) if organization: query = query.filter(User.organization == organization) if exclude_admins: query = query.filter(User.role != ADMIN) query = query.order_by(User.name) limited_query = query.limit(limit).offset(offset) rows = limited_query.all() users = [row.to_dict() for row in rows] total_records = query.count() return (users, total_records) def delete_user(user_id, session=None): """Delete a user. Args: user_id (int): the user id. Returns: dict: containing the deleted user dict. """ if session: return _delete_user(user_id, session) with mysql.pod_db_session() as session: return _delete_user(user_id, session) def _delete_user(user_id, session): user = session.query(User).get(user_id) if not user: raise exc.OwsError.not_found(error.ERROR_MESSAGE_USER_NOT_FOUND) if not user.active: raise exc.OwsError.not_found(error.ERROR_MESSAGE_USER_IS_DELETED) user.active = False return user.to_dict() def hard_delete_user(user_id): """Remove a user from the db. Args: user_id (str): the user id. Returns: dict: containing the user dict. """ with mysql.pod_db_session() as session: session.query(User).filter( User.id == user_id ).delete() def update_user(user_id, data): """Update a user. Args: user_id (str): the user id. data (dict): the update dictionary Returns: dict: containing the user dict. """ with mysql.pod_db_session() as session: network_ids = data.pop('network_ids', []) podcast_ids = data.pop('podcast_ids', []) show_family_ids = data.pop('show_family_ids', None) query = session.query(User).filter(User.id == user_id) user = query.first() if not user: raise exc.OwsError.not_found(error.ERROR_MESSAGE_USER_NOT_FOUND) if user.role == ADMIN and data.get('role'): raise exc.OwsError.bad_request(error.ERROR_INCORRECT_ROLE) data['updated_by'] = api_utils.get_user_id() query.update(data) result = query.first() if network_ids: result.networks = session.query(network_model.Network).filter( network_model.Network.id.in_(network_ids)).all() else: result.networks = [] if feature_flag_utils.get_feature_flag(FEATURE_PODCAST_IA_RESTRUCTURE): if show_family_ids: result.show_families = session.query(show_family_model.ShowFamily).filter( show_family_model.ShowFamily.id.in_(show_family_ids)).all() else: result.show_families = [] else: if podcast_ids: result.podcasts = session.query(podcast_model.Podcast).filter( podcast_model.Podcast.id.in_(podcast_ids)).all() else: result.podcasts = [] return result.to_dict() def update_user_last_login(user_id, data): """Update user's last login datetime. Args: user_id (str): the user id. data (dict): the update dictionary Returns: dict: containing the user dict. """ with mysql.pod_db_session() as session: query = session.query(User).filter(User.id == user_id) user = query.first() if not user: raise exc.OwsError.not_found(error.ERROR_MESSAGE_USER_NOT_FOUND) data['updated_by'] = user_id query.update(data) return user.to_dict() def _get_user_by_uuid(user_id, session): """Check if user exists in db. Args: uuid (str): id """ filters = [ User.uuid == user_id, User.active.is_(True) ] return session.query(User).filter(*filters).first() def users_for_network_ids(network_ids): """Get users for network id. Args: network_id (int): the network id. Returns: list: containing the user dicts. """ with mysql.pod_db_session(read_only=True) as session: results = session.query(User)\ .filter(User.networks.any(network_model.Network.id.in_(network_ids))) \ .filter(User.active.is_(True)) \ .all() return { 'items': [result.to_dict() for result in results] } def users_for_podcast_and_network_ids(podcast_ids, network_ids): """Get users for a specific podcast. Args: network_id (int): the network id. podcast_id (int): the podcast ids Returns: list: containing the user dicts. """ with mysql.pod_db_session(read_only=True) as session: results = session.query(User)\ .filter(sqlalchemy.or_( User.networks.any(network_model.Network.id.in_(network_ids)), User.podcasts.any(podcast_model.Podcast.id.in_(podcast_ids)))) \ .filter(User.active.is_(True)) \ .all() return { 'items': [result.to_dict() for result in results] } def users_for_show_family_and_network_ids(show_family_ids, network_ids): """Get users belonging to show families and network ids. Args: network_ids (list): the network ids. show_family_ids (list): the show family ids Returns: list: containing the user dicts. """ with mysql.pod_db_session(read_only=True) as session: results = session.query(User)\ .filter(sqlalchemy.or_( User.networks.any(network_model.Network.id.in_(network_ids)), User.show_families.any(show_family_model.ShowFamily.id.in_(show_family_ids)))) \ .filter(User.active.is_(True)) \ .all() return { 'items': [result.to_dict() for result in results] } def get_users_by_ids(ids, is_only_users=False): """Get users by ids. Args: ids (list): ids to fetch Returns: dict: containing the user dicts. """ with mysql.pod_db_session(read_only=True) as session: rows = session.query(User).filter(User.id.in_(ids)) \ .filter(User.active.is_(True)).all() items = [row.to_dict(is_only_users) for row in rows] return {'items': items} def all_access_users(): """Get users with all access. Returns: dict: containing the user dicts. """ with mysql.pod_db_session(read_only=True) as session: rows = session.query(User).filter(User.all_networks.is_(True)) \ .filter(User.active.is_(True)).all() items = [row.to_dict() for row in rows] return {'items': items} def get_user_favorites(user_id): """Get user favorites. Args: id (int): user Returns: dict: favorites """ with mysql.pod_db_session(read_only=True) as session: user = session.query(User).filter(User.id == user_id) \ .filter(User.active.is_(True)).first() podcasts = [{'podcast_id': p.id} for p in user.favorite_podcasts] user_chart_favorites = session.query(UserChartFavorite).filter(UserChartFavorite.user_id == user_id) charts = [ {'store': c.store, 'category': c.category, 'country': c.country, 'chart_type': c.chart_type} for c in user_chart_favorites.all() ] return { 'podcasts': podcasts, 'charts': charts } def toggle_user_podcast_favorite(user_id, podcast_id): """Toggle user podcast favorite. Args: id (int): user podcast_id (int): podcast id Returns: list: favorites """ with mysql.pod_db_session() as session: user = session.query(User).filter(User.id == user_id) \ .filter(User.active.is_(True)).first() podcast = session.query(podcast_model.Podcast).get(podcast_id) if podcast in user.favorite_podcasts: user.favorite_podcasts.remove(podcast) else: user.favorite_podcasts.append(podcast) return get_user_favorites(user_id) def toggle_user_chart_favorite(user_id, store, category, country, chart_type): """Toggle user chart favorite.""" with mysql.pod_db_session() as session: user_chart_favorite = session.query(UserChartFavorite) \ .filter(UserChartFavorite.user_id == user_id) \ .filter(UserChartFavorite.store == store) \ .filter(UserChartFavorite.category == category) \ .filter(UserChartFavorite.country == country) \ .filter(UserChartFavorite.chart_type == chart_type) if user_chart_favorite.first(): user_chart_favorite.delete() else: user = UserChartFavorite( user_id=user_id, store=store, category=category, country=country, chart_type=chart_type) session.add(user) return get_user_favorites(user_id) def get_user_by_email(email): """Check if user exists in db. Args: email (str): email """ with mysql.pod_db_session(read_only=True) as session: user = _get_user_by_email(email, session) return user.to_dict() if user else None def _get_user_by_email(email, session): """Get single user object.""" return session.query(User).filter(User.email == email) \ .filter(User.active.is_(True)).first() def delete_favorite_podcast_by_podcast_id(podcast_id, session): """Delete favorited podcast by podcast id.""" delete_stmt = user_podcast_favorite_table.delete().where( user_podcast_favorite_table.c.podcast_id == podcast_id) session.execute(delete_stmt)