Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use handle_purge method in avocado

Best Python code snippet using avocado_python

cache_toprequesters.py

Source:cache_toprequesters.py

...60            pass  # do nothing, go straight to print_status61        else:62            pid = Pid(self.options.get('pidfile'), True)63            self.prepare_options()  # sets self.reports and prints errors is any64            self.reports and (self.handle_purge() or self.handle_cache())65            pid.remove_pid()66        self.print_status()67    def prepare_options(self):68        """Sets self.reports with an array of data to process or None.69        Prints any errors occurring.70        """71        self.reports = None72        if self.options.get('all'):73            self.reports = ToprequestersReport.values74        else:75            report_type = self.options.get('report-type')76            if report_type in ToprequestersReport.values:77                self.reports = [report_type]78            else:79                if report_type is None:80                    log.info('Please specify "--report-type=<type>" or '81                        '"--all".')82                else:83                    log.info('Unknown report type: "{0}".'.format(84                        report_type))85    def handle_cache(self):86        """Handles the primary cache function of this command."""87        log.info('Caching reports: {0}.'.format(self.reports))88        for report_type in self.reports:89            self.__cache_report(report_type)90    def __cache_report(self, report_type):91        """Evaluates the report and stores it under correct cache key."""92        display_name = ToprequestersReport.display_names[report_type]93        if ToprequestersReport.is_cached(report_type):94            if self.options['force']:95                log.info('Recalculating "{0}" toprequesters report.'.format(96                    display_name))97            else:98                log.info('"{0}" toprequesters still in cache, use --force flag'99                    ' to rebuild anyway.'.format(display_name))100                return False101        else:102            log.info(('"{0}" toprequesters report missing, recalculating.'103                ).format(display_name))104        start_date = now()105        start_time = time.time()106        days = int(self.options['days'])107        data = ToprequestersReport.REPORT_FUNCTION[report_type](days)108        elapsed = time.time() - start_time109        log.info('Toprequesters report "{0}" generated in: {1}s.'.format(110            display_name, elapsed))111        # too often we get no information on the success of caching112        if not data:113            log.warning('Data returned by report function is {0}!'.format(data))114        else:115            meta = {116                'days': days,117                'to': start_date.date(),118                'from': (start_date - datetime.timedelta(days=days)).date(),119                'start_time': start_date,120                'elapsed': elapsed,121            }122            ToprequestersReport.store(report_type, data, meta)123            if not ToprequestersReport.is_cached(report_type):124                log.warning('Cache error - data could not be fetched!')125        return True126    def handle_purge(self):127        """Handles purge option. Removes all data if --all was used or specified128        --report-type=<id>.129        """130        if self.options['purge']:131            log.info("Removing data for {0}.".format(self.reports))132            for r in self.reports:133                ToprequestersReport.purge(r)134            self.print_status()135        return self.options['purge']136    def print_status(self):137        """Shows report availability report."""138        log.info("Available report types: \n" +...

wsgi.py

Source:wsgi.py

...21    path_and_query = request_uri.lstrip('/')22    if path_and_query is "":23        return do_404(environ, start_response, "Not Found", DEBUG)24    if environ.get('REQUEST_METHOD', 'GET') == 'PURGE':25        return handle_purge(environ, start_response)26    # Acquire lockfile27    lock = '/tmp/%s' % sha1(path_and_query).hexdigest()28    if os.path.isfile(lock):29        return do_404(environ, start_response, "File is being processed", DEBUG)30    with lock_file(lock):31        try:32            server = environ['SERVER_NAME']33            new_file = Transmogrify(path_and_query, server)34            new_file.save()35        except Http404 as e:36            return do_404(environ, start_response, e.message, DEBUG)37        return do_redirect(environ, start_response, request_uri)38if __name__ == '__main__':39    from wsgiref.simple_server import make_server...

purge.py

Source:purge.py

1from pyrogram.handlers import MessageHandler2from django.conf import settings3from pyrogram import filters4async def handle_purge(client, msg):5    if msg.from_user.id != settings.BOT_MASTER:6        await msg.delete()7        return False8    if not msg.reply_to_message:9        await msg.delete()10        return False11    target_msg = msg.reply_to_message12    purge_list = list(range(target_msg.id, msg.id+1))13    reason = None14    if len(msg.command) > 1:15        reason = ' '.join(msg.command[1:])16    print(reason)17    try:18        await client.delete_messages(msg.chat.id,...

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.