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

How to use delete_streams method in localstack

Best Python code snippet using localstack_python

analyze_thread.py

Source:analyze_thread.py

1'''2Created on 2011-12-173@author: dista4'''5import threading6import asyncore7from media_server_admin_queryer import MediaServerAdminQueryer8from analyzer import Analyzer9import time10import logger11from model.stream_model import StreamModel12from model.sample_model import SampleModel13import MySQLdb14class AnalyzeThread(threading.Thread):15    def __init__(self):16        threading.Thread.__init__(self)17        self.logger = logger.get_logger()18        self.obs = None19        self.sleep_time = 320        self.analyzer = Analyzer()21        self.obs_lock = threading.Lock()22        self.stream_model = StreamModel() 23        self.sample_model = SampleModel()24    25    def notify_obs_changed(self, new_obs):26        self.obs_lock.acquire()27        self.obs = new_obs28        self.obs_lock.release()29    def _get_obs(self):30        ret = None31        self.obs_lock.acquire()32        ret = self.obs33        self.obs_lock.release()34        return ret 35    def _get_map_stream_ids(self, _map):36        return [msq.stream_id for msq in _map.items]37    38    def run(self):39        while True:40            all_streams = self._get_obs()41            streams_in_db = self.stream_model.get_all()42            if all_streams == None:43                all_streams = streams_in_db44            45            sended_requests = []46            _map = {}47            for stream in all_streams:48                sended_requests.append(MediaServerAdminQueryer(self.analyzer, stream['stream_id'], stream['sample_interface'], _map))49            asyncore.loop(map = _map)50            51            delete_streams, update_streams, add_streams = self._get_changed_streams(streams_in_db, all_streams)52            try:53                self._update_db(delete_streams, update_streams, add_streams, sended_requests)54            except MySQLdb.Error, e:55                self.logger.warn("update_db failed")56            time.sleep(self.sleep_time)57    def _get_changed_streams(self, streams_before_run, streams_after_run):58        before_ids = set([stream['stream_id'] for stream in streams_before_run])59        after_ids = set([stream['stream_id'] for stream in streams_after_run])60        delete_ids, update_ids, add_ids =  before_ids - after_ids, before_ids & after_ids, after_ids - before_ids61        62        delete_streams = [stream for stream in streams_before_run if (stream['stream_id'] in delete_ids)]63        update_streams_after = [stream for stream in streams_after_run if (stream['stream_id'] in update_ids)]64        update_streams_before = [stream for stream in streams_before_run if (stream['stream_id'] in update_ids)]65        add_streams  = [stream for stream in streams_after_run if (stream['stream_id'] in add_ids)]66        return (delete_streams, self._get_update_streams(update_streams_after, update_streams_before), add_streams)67    def _get_update_streams(self, update_streams_after, update_streams_before):68        ret = []69        for s1 in update_streams_after:70            for s2 in update_streams_before:71                if s2['stream_id'] == s1['stream_id']:72                    if s1['sample_interface'] != s2['sample_interface'] or s1['unify_name'] != s2['unify_name']:73                        ret.append(s1)74                    break75        return ret76            77    def _update_db(self, delete_streams, update_streams, add_streams, sended_requests):78        '''79        HERE WE UPDATE DB!! AND ONLY HERE WE DO THAT80        '''81        mms_streams = self.stream_model.get_by_stream_ids([stream['stream_id'] for stream in delete_streams])82        mms_streams_ids = [stream['id'] for stream in mms_streams]83        self.sample_model.delete_by_stream_ids(mms_streams_ids)84        self.stream_model.delete_by_ids(mms_streams_ids)85        for stream in update_streams:86            self.stream_model.update_by_stream_id(stream)87        self.stream_model.add(add_streams)88        for request in sended_requests:89            analyze_result = request.analyze_result90            stream = analyze_result['cal_data']91            stream['stream_id'] = request.stream_id92            stream['score'] = analyze_result['score']93            stream['score_level'] = analyze_result['score_level']94            stream['score_upstream'] = analyze_result['score_detail']['upstream_value']95            stream['score_level_upstream'] = analyze_result['score_level_detail']['upstream']96            stream['score_live_delay'] = analyze_result['score_detail']['live_delay_value']97            stream['score_level_live_delay'] = analyze_result['score_level_detail']['live_delay']98            stream['last_sample_time'] = request.sample_time.strftime("%Y-%m-%d %H:%M:%S")99            100            self.stream_model.update_by_stream_id(stream)101            102            stream_in_db = self.stream_model.get_by_stream_id(request.stream_id)103            analyze_result['sample']['mms_stream_id'] = stream_in_db[0]['id']104            analyze_result['sample']['sample_time'] = analyze_result['sample']['sample_time'].strftime("%Y-%m-%d %H:%M:%S")...

signals.py

Source:signals.py

1from django import dispatch2from django.db.models import signals as django_signals3from nodewatcher.core import models as core_models4from . import tasks5@dispatch.receiver(django_signals.post_delete, sender=core_models.Node)6def datastream_node_removed(sender, instance, **kwargs):7    """8    Remove all streams when a node gets removed.9    """10    tasks.delete_streams.delay({'node': instance.pk})11# In case we have the frontend module installed, we also subscribe to its12# reset signal that gets called when a user requests a node's data to be reset13try:14    from nodewatcher.modules.frontend.editor import signals as editor_signals15    @dispatch.receiver(editor_signals.reset_node)16    def datastream_node_reset(sender, request, node, **kwargs):17        """18        Remove all streams when a user requests the node to be reset.19        """20        tasks.delete_streams.delay({'node': node.pk})21except ImportError:...

tasks.py

Source:tasks.py

...7    as some backends need this to be executed periodically.8    """9    datastream.downsample_streams()10@celery.app.task(queue='monitor')11def delete_streams(tags):12    """13    Deletes datastream streams matching ``tags``.14    """...

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.