KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use handle_connection method in localstack

Best Python code snippet using localstack_python

db_shared.py

Source:db_shared.py

1# -*- coding: utf-8 -*-2"""3    Copyright (C) 2017 Sebastian Golasch (plugin.video.netflix)4    Copyright (C) 2019 Stefano Gottardo - @CastagnaIT (original implementation module)5    Sharable database access and functions6    SPDX-License-Identifier: MIT7    See LICENSES/MIT.md for more information.8"""9from __future__ import absolute_import, division, unicode_literals10from datetime import datetime11import resources.lib.common as common12import resources.lib.database.db_base_mysql as db_base_mysql13import resources.lib.database.db_base_sqlite as db_base_sqlite14import resources.lib.database.db_utils as db_utils15def get_shareddb_class(use_mysql=False):16    # Dynamically sets the inherit class17    base_class = db_base_mysql.MySQLDatabase if use_mysql else db_base_sqlite.SQLiteDatabase18    class NFSharedDatabase(base_class):19        def __init__(self):20            if use_mysql:21                super(NFSharedDatabase, self).__init__()22            else:23                super(NFSharedDatabase, self).__init__(db_utils.SHARED_DB_FILENAME)24        def get_value(self, key, default_value=None, table=db_utils.TABLE_SHARED_APP_CONF, data_type=None):  # pylint: disable=useless-super-delegation25            return super(NFSharedDatabase, self).get_value(key, default_value, table, data_type)26        def get_values(self, key, default_value=None, table=db_utils.TABLE_SHARED_APP_CONF):  # pylint: disable=useless-super-delegation27            return super(NFSharedDatabase, self).get_values(key, default_value, table)28        def set_value(self, key, value, table=db_utils.TABLE_SHARED_APP_CONF):  # pylint: disable=useless-super-delegation29            super(NFSharedDatabase, self).set_value(key, value, table)30        def delete_key(self, key, table=db_utils.TABLE_SHARED_APP_CONF):  # pylint: disable=useless-super-delegation31            super(NFSharedDatabase, self).delete_key(key, table)32        @db_base_mysql.handle_connection33        @db_base_sqlite.handle_connection34        def set_profile(self, guid, sort_order):35            """Update or Insert a profile"""36            # Update or insert approach,37            # if there is no updated row then insert new one (no id changes)38            if self.is_mysql_database:39                query = db_utils.mysql_insert_or_update('profiles', ['Guid'], ['SortOrder'])40                self._execute_non_query(query, (guid, sort_order), multi=True)41            else:42                data = db_utils.sql_filtered_update('profiles',43                                                    ['SortOrder'],44                                                    ['Guid'],45                                                    [sort_order, guid])46                cur = self._execute_query(data[0], data[1])47                if cur.rowcount == 0:48                    data = db_utils.sql_filtered_insert('profiles',49                                                        ['Guid', 'SortOrder'],50                                                        [guid, sort_order])51                    self._execute_non_query(data[0], data[1])52        @db_base_mysql.handle_connection53        @db_base_sqlite.handle_connection54        def delete_profile(self, guid):55            query = 'DELETE FROM profiles WHERE Guid = ?'56            self._execute_non_query(query, (guid,))57        @db_base_mysql.handle_connection58        @db_base_sqlite.handle_connection59        def get_movie_filepath(self, movieid, default_value=None):60            """Get movie filepath for given id"""61            query = 'SELECT FilePath FROM video_lib_movies WHERE MovieID = ?'62            cur = self._execute_query(query, (movieid,))63            result = cur.fetchone()64            return result[0] if result else default_value65        @db_base_mysql.handle_connection66        @db_base_sqlite.handle_connection67        def get_episode_filepath(self, tvshowid, seasonid, episodeid, default_value=None):68            """Get movie filepath for given id"""69            query =\70                ('SELECT FilePath FROM video_lib_episodes '71                 'INNER JOIN video_lib_seasons '72                 'ON video_lib_episodes.SeasonID = video_lib_seasons.SeasonID '73                 'WHERE video_lib_seasons.TvShowID = ? AND '74                 'video_lib_seasons.SeasonID = ? AND '75                 'video_lib_episodes.EpisodeID = ?')76            cur = self._execute_query(query, (tvshowid, seasonid, episodeid))77            result = cur.fetchone()78            return result[0] if result is not None else default_value79        @db_base_mysql.handle_connection80        @db_base_sqlite.handle_connection81        def get_all_episodes_ids_and_filepath_from_tvshow(self, tvshowid):82            """Get all episodes IDs and filepaths for given id"""83            cur = self.get_cursor_for_dict_results()84            query =\85                ('SELECT video_lib_episodes.FilePath, video_lib_seasons.TvShowID, '86                 'video_lib_episodes.SeasonID, video_lib_episodes.EpisodeID '87                 'FROM video_lib_episodes '88                 'INNER JOIN video_lib_seasons '89                 'ON video_lib_episodes.SeasonID = video_lib_seasons.SeasonID '90                 'WHERE video_lib_seasons.TvShowID = ?')91            cur = self._execute_query(query, (tvshowid,), cur)92            return cur.fetchall()93        @db_base_mysql.handle_connection94        @db_base_sqlite.handle_connection95        def get_all_episodes_ids_and_filepath_from_season(self, tvshowid, seasonid):96            """Get all episodes IDs and filepaths for given id"""97            cur = self.get_cursor_for_dict_results()98            query =\99                ('SELECT video_lib_episodes.FilePath, video_lib_seasons.TvShowID, '100                 'video_lib_episodes.SeasonID, video_lib_episodes.EpisodeID '101                 'FROM video_lib_episodes '102                 'INNER JOIN video_lib_seasons '103                 'ON video_lib_episodes.SeasonID = video_lib_seasons.SeasonID '104                 'WHERE video_lib_seasons.TvShowID = ? AND '105                 'video_lib_seasons.SeasonID = ?')106            cur = self._execute_query(query, (tvshowid, seasonid), cur)107            return cur.fetchall()108        @db_base_mysql.handle_connection109        @db_base_sqlite.handle_connection110        def get_random_episode_filepath_from_tvshow(self, tvshowid, default_value=None):111            """Get random episode filepath of a show of a given id"""112            rand_func_name = 'RAND()' if self.is_mysql_database else 'RANDOM()'113            query =\114                ('SELECT FilePath FROM video_lib_episodes '115                 'INNER JOIN video_lib_seasons '116                 'ON video_lib_episodes.SeasonID = video_lib_seasons.SeasonID '117                 'WHERE video_lib_seasons.TvShowID = ? '118                 'ORDER BY {} LIMIT 1').format(rand_func_name)119            cur = self._execute_query(query, (tvshowid,))120            result = cur.fetchone()121            return result[0] if result is not None else default_value122        @db_base_mysql.handle_connection123        @db_base_sqlite.handle_connection124        def get_random_episode_filepath_from_season(self, tvshowid, seasonid, default_value=None):125            """Get random episode filepath of a show of a given id"""126            rand_func_name = 'RAND()' if self.is_mysql_database else 'RANDOM()'127            query =\128                ('SELECT FilePath FROM video_lib_episodes '129                 'INNER JOIN video_lib_seasons '130                 'ON video_lib_episodes.SeasonID = video_lib_seasons.SeasonID '131                 'WHERE video_lib_seasons.TvShowID = ? AND video_lib_seasons.SeasonID = ? '132                 'ORDER BY {} LIMIT 1').format(rand_func_name)133            cur = self._execute_query(query, (tvshowid, seasonid))134            result = cur.fetchone()135            return result[0] if result is not None else default_value136        @db_base_mysql.handle_connection137        @db_base_sqlite.handle_connection138        def get_all_video_id_list(self):139            """Get all the ids of movies and tvshows contained in the library"""140            cur = self.get_cursor_for_list_results()141            query = ('SELECT MovieID FROM video_lib_movies '142                     'UNION '143                     'SELECT TvShowID FROM video_lib_tvshows')144            cur = self._execute_query(query, cursor=cur)145            return self.return_rows_as_list(cur)146        @db_base_mysql.handle_connection147        @db_base_sqlite.handle_connection148        def get_tvshows_id_list(self, enum_vid_prop=None, prop_value=None):149            """150            Get all the ids of tvshows contained in the library151            :param enum_vid_prop: Optional: use db_utils.VidLibProp152            :param prop_value: Optional: value as filter153            :return: list of tvshows ids154            """155            cur = self.get_cursor_for_list_results()156            if enum_vid_prop and prop_value:157                query = ('SELECT TvShowID FROM video_lib_tvshows '158                         'WHERE ' + enum_vid_prop + ' = ?')159                cur = self._execute_query(query, (str(prop_value),), cur)160            else:161                query = 'SELECT TvShowID FROM video_lib_tvshows'162                cur = self._execute_query(query, cursor=cur)163            return self.return_rows_as_list(cur)164        @db_base_mysql.handle_connection165        @db_base_sqlite.handle_connection166        def get_movies_id_list(self):167            """Get all the ids of movies contained in the library"""168            cur = self.get_cursor_for_list_results()169            query = 'SELECT MovieID FROM video_lib_movies'170            cur = self._execute_query(query, cursor=cur)171            return self.return_rows_as_list(cur)172        @db_base_mysql.handle_connection173        @db_base_sqlite.handle_connection174        def movie_id_exists(self, movieid):175            """Return True if a movie id exists"""176            query = 'SELECT EXISTS(SELECT 1 FROM video_lib_movies WHERE MovieID = ?)'177            cur = self._execute_query(query, (movieid,))178            return bool(cur.fetchone()[0])179        @db_base_mysql.handle_connection180        @db_base_sqlite.handle_connection181        def tvshow_id_exists(self, tvshowid):182            """Return True if a tvshow id exists"""183            query = 'SELECT EXISTS(SELECT 1 FROM video_lib_tvshows WHERE TvShowID = ?)'184            cur = self._execute_query(query, (tvshowid,))185            return bool(cur.fetchone()[0])186        @db_base_mysql.handle_connection187        @db_base_sqlite.handle_connection188        def season_id_exists(self, tvshowid, seasonid):189            """Return True if a tvshow season id exists"""190            query =\191                ('SELECT EXISTS('192                 'SELECT 1 FROM video_lib_seasons '193                 'INNER JOIN video_lib_tvshows '194                 'ON video_lib_seasons.TvShowID = video_lib_tvshows.TvShowID '195                 'WHERE video_lib_tvshows.TvShowID = ? AND video_lib_seasons.SeasonID = ?)')196            cur = self._execute_query(query, (tvshowid, seasonid))197            return bool(cur.fetchone()[0])198        @db_base_mysql.handle_connection199        @db_base_sqlite.handle_connection200        def episode_id_exists(self, tvshowid, seasonid, episodeid):201            """Return True if a tvshow episode id exists"""202            query =\203                ('SELECT EXISTS('204                 'SELECT 1 FROM video_lib_episodes '205                 'INNER JOIN video_lib_seasons '206                 'ON video_lib_episodes.SeasonID = video_lib_seasons.SeasonID '207                 'INNER JOIN video_lib_tvshows '208                 'ON video_lib_seasons.TvShowID = video_lib_tvshows.TvShowID '209                 'WHERE video_lib_tvshows.TvShowID = ? AND '210                 'video_lib_seasons.SeasonID = ? AND '211                 'video_lib_episodes.EpisodeID = ?)')212            cur = self._execute_query(query, (tvshowid, seasonid, episodeid))213            return bool(cur.fetchone()[0])214        @db_base_mysql.handle_connection215        @db_base_sqlite.handle_connection216        def set_movie(self, movieid, file_path, nfo_export):217            """Update or insert a movie"""218            # Update or insert approach, if there is no updated row then insert new one219            if self.is_mysql_database:220                query = db_utils.mysql_insert_or_update('video_lib_movies', ['MovieID'],221                                                        ['FilePath', 'NfoExport'])222                self._execute_non_query(query, (movieid, file_path, str(nfo_export)), multi=True)223            else:224                update_query = ('UPDATE video_lib_movies SET FilePath = ?, NfoExport = ? '225                                'WHERE MovieID = ?')226                cur = self._execute_query(update_query, (file_path, str(nfo_export), movieid))227                if cur.rowcount == 0:228                    insert_query = ('INSERT INTO video_lib_movies (MovieID, FilePath, NfoExport) '229                                    'VALUES (?, ?, ?)')230                    self._execute_non_query(insert_query, (movieid, file_path, str(nfo_export)))231        @db_base_mysql.handle_connection232        @db_base_sqlite.handle_connection233        def set_tvshow(self, tvshowid, nfo_export, exclude_update):234            """Update or insert a tvshow"""235            # Update or insert approach, if there is no updated row then insert new one236            if self.is_mysql_database:237                query = db_utils.mysql_insert_or_update('video_lib_tvshows', ['TvShowID'],238                                                        ['ExcludeUpdate', 'NfoExport'])239                self._execute_non_query(query, (tvshowid, str(exclude_update), str(nfo_export)),240                                        multi=True)241            else:242                update_query = ('UPDATE video_lib_tvshows SET NfoExport = ?, ExcludeUpdate = ? '243                                'WHERE TvShowID = ?')244                cur = self._execute_query(update_query, (str(nfo_export),245                                                         str(exclude_update), tvshowid))246                if cur.rowcount == 0:247                    insert_query = \248                        ('INSERT INTO video_lib_tvshows (TvShowID, NfoExport, ExcludeUpdate) '249                         'VALUES (?, ?, ?)')250                    self._execute_non_query(insert_query, (tvshowid,251                                                           str(nfo_export),252                                                           str(exclude_update)))253        @db_base_mysql.handle_connection254        @db_base_sqlite.handle_connection255        def insert_season(self, tvshowid, seasonid):256            """Insert a season if not exists"""257            if not self.season_id_exists(tvshowid, seasonid):258                insert_query = ('INSERT INTO video_lib_seasons (TvShowID, SeasonID) '259                                'VALUES (?, ?)')260                self._execute_non_query(insert_query, (tvshowid, seasonid))261        @db_base_mysql.handle_connection262        @db_base_sqlite.handle_connection263        def insert_episode(self, tvshowid, seasonid, episodeid, file_path):264            """Insert a episode if not exists"""265            if not self.episode_id_exists(tvshowid, seasonid, episodeid):266                insert_query = ('INSERT INTO video_lib_episodes (SeasonID, EpisodeID, FilePath) '267                                'VALUES (?, ?, ?)')268                self._execute_non_query(insert_query, (seasonid, episodeid, file_path))269        @db_base_mysql.handle_connection270        @db_base_sqlite.handle_connection271        def delete_movie(self, movieid):272            """Delete a movie from database"""273            query = 'DELETE FROM video_lib_movies WHERE MovieID = ?'274            self._execute_query(query, (movieid,))275        @db_base_mysql.handle_connection276        @db_base_sqlite.handle_connection277        def delete_tvshow(self, tvshowid):278            """Delete a tvshow from database"""279            query = 'DELETE FROM video_lib_tvshows WHERE TvShowID = ?'280            self._execute_query(query, (tvshowid,))281        @db_base_mysql.handle_connection282        @db_base_sqlite.handle_connection283        def delete_season(self, tvshowid, seasonid):284            """Delete a season from database"""285            query = 'DELETE FROM video_lib_seasons WHERE TvShowID = ? AND SeasonID = ?'286            self._execute_query(query, (tvshowid, seasonid))287            # if there are no other seasons, delete the tvshow288            query = 'SELECT EXISTS(SELECT 1 FROM video_lib_seasons WHERE TvShowID = ?)'289            cur = self._execute_query(query, (tvshowid,))290            if not bool(cur.fetchone()[0]):291                self.delete_tvshow(tvshowid)292        @db_base_mysql.handle_connection293        @db_base_sqlite.handle_connection294        def delete_episode(self, tvshowid, seasonid, episodeid):295            """Delete a episode from database"""296            query = 'DELETE FROM video_lib_episodes WHERE SeasonID = ? AND EpisodeID = ?'297            self._execute_query(query, (seasonid, episodeid))298            # if there are no other episodes, delete the season299            query = 'SELECT EXISTS(SELECT 1 FROM video_lib_episodes WHERE SeasonID = ?)'300            cur = self._execute_query(query, (seasonid,))301            if not bool(cur.fetchone()[0]):302                self.delete_season(tvshowid, seasonid)303        @db_base_mysql.handle_connection304        @db_base_sqlite.handle_connection305        def get_tvshow_property(self, tvshowid, enum_vid_prop, default_value=None, data_type=None):306            """307            Read the value of the specified property308            :param tvshowid: id of tvshow309            :param enum_vid_prop: Use a enum value of db_utils.VidLibProp310            :param default_value: When key do not exist return this default value311            :param data_type: OPTIONAL Used to set data type conversion only when default_value is None312            :return: the property value313            """314            query = 'SELECT ' + enum_vid_prop + ' FROM video_lib_tvshows WHERE TvShowID = ?'315            cur = self._execute_query(query, (tvshowid,))316            result = cur.fetchone()317            if default_value is not None:318                data_type = type(default_value)319            elif data_type is None:320                data_type = str321            return common.convert_from_string(result[0], data_type) \322                if result is not None else default_value323        @db_base_mysql.handle_connection324        @db_base_sqlite.handle_connection325        def set_tvshow_property(self, tvshowid, enum_vid_prop, value):326            update_query = ('UPDATE video_lib_tvshows '327                            'SET ' + enum_vid_prop + ' = ? WHERE TvShowID = ?')328            value = common.convert_to_string(value)329            self._execute_query(update_query, (value, tvshowid))330        @db_base_mysql.handle_connection331        @db_base_sqlite.handle_connection332        def get_stream_continuity(self, profile_guid, videoid, default_value=None, data_type=None):333            """Get stream continuity value of a given id stored to current profile"""334            query = 'SELECT Value FROM stream_continuity WHERE ProfileGuid = ? AND VideoID = ?'335            cur = self._execute_query(query, (profile_guid, videoid))336            result = cur.fetchone()337            if default_value is not None:338                data_type = type(default_value)339            elif data_type is None:340                data_type = str341            return common.convert_from_string(result[0], data_type) \342                if result is not None else default_value343        @db_base_mysql.handle_connection344        @db_base_sqlite.handle_connection345        def set_stream_continuity(self, profile_guid, videoid, value):346            """Update or insert a stream continuity value to current profile"""347            # Update or insert approach, if there is no updated row then insert new one348            value = common.convert_to_string(value)349            date_last_modified = common.convert_to_string(datetime.now())350            if self.is_mysql_database:351                query = db_utils.mysql_insert_or_update('stream_continuity',352                                                        ['ProfileGuid', 'VideoID'],353                                                        ['Value', 'DateLastModified'])354                self._execute_non_query(query, (profile_guid, videoid, value, date_last_modified),355                                        multi=True)356            else:357                update_query = ('UPDATE stream_continuity '358                                'SET Value = ?, DateLastModified = ? '359                                'WHERE ProfileGuid = ? AND VideoID = ?')360                cur = self._execute_query(update_query, (value, date_last_modified,361                                                         profile_guid, videoid))362                if cur.rowcount == 0:363                    insert_query = ('INSERT INTO stream_continuity '364                                    '(ProfileGuid, VideoID, Value, DateLastModified) '365                                    'VALUES (?, ?, ?, ?)')366                    self._execute_non_query(insert_query, (profile_guid, videoid,367                                                           value, date_last_modified))368        @db_base_mysql.handle_connection369        @db_base_sqlite.handle_connection370        def purge_library(self):371            """Delete all records from library tables"""372            query = 'DELETE FROM video_lib_movies'373            self._execute_non_query(query)374            query = 'DELETE FROM video_lib_episodes'375            self._execute_non_query(query)376            query = 'DELETE FROM video_lib_seasons'377            self._execute_non_query(query)378            query = 'DELETE FROM video_lib_tvshows'379            self._execute_non_query(query)...

functions.py

Source:functions.py

...10except requests.exceptions.ConnectionError:11    DEVICE_ID = None12except IndexError:13    DEVICE_ID = None14def handle_connection(func):15    """Decorator that takes in a function as argument and checks if the function gives an error"""16    @wraps(func)17    def decorated(*args, **kwargs):18        try:19            return func(*args, **kwargs)20        except requests.exceptions.ConnectionError:21            return {'status': 'error',22                    'message': 'ConnectionError, no internet'}23        except requests.exceptions.ReadTimeout:24            return {'status': 'error',25                    'message': 'ReadTimeOut, could not send request'}26        except spotipy.SpotifyException:27            global spotifyHandler28            spotifyHandler = connect()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.