How to use gramex - 10 common examples

def check(reload):
            result, reloaded = gramex.cache.open(
                path, 'template', _reload_status=True, autoescape=None)
            eq_(reloaded, reload)
            ok_(isinstance(result, Template))
            first_line = result.generate(name='高').decode('utf-8').split('\n')[0]
            eq_(first_line, '1: name=高')

def check_edit(self, method, source, args, count):
        # Edits the correct count of records, returns empty value and saves to file
        target = self.copy_file('sales.xlsx', 'sales-edits.xlsx')
        self.call('xlsx-' + source, args, method, {'Count-Data': str(count)})
        result = gramex.cache.open(target)
        # Check result. TODO: check that the values are correctly added
        if method == 'delete':
            eq_(len(result), len(self.sales) - count)
        elif method == 'post':
            eq_(len(result), len(self.sales) + count)
        elif method == 'put':
            eq_(len(result), len(self.sales))

        target = os.path.join(folder, 'formhandler-edits.db')
        dbutils.sqlite_create_db(target, sales=self.sales)
        tempfiles[target] = target
        self.call('sqlite-' + source, args, method, {'Count-Data': str(count)})
        # Check result. TODO: check that the values are correctly added
        con = sqlite3.connect(target)
        result = pd.read_sql('SELECT * FROM sales', con)
        # TODO: check that the values are correctly added

def test_filtercols_limit(self):
        # ?_c=District&_limit=200 returns 200 values. (_limit= defaults to 100 values)
        cols = ['District']
        limit = 200
        args = {'_c': cols, '_limit': [limit]}
        result = gramex.data.filtercols(args=args, **self.urls['census'])
        self.check_keys(result, cols)
        for key, val in result.items():
            eqframe(val, self.unique_of(self.census, key).head(limit))

'path': 'search/tweets.json', 'data': search},
        }
        # Fetch the tweets upfront, asynchronously
        futures = [async_fetch(name, **kwargs) for name, kwargs in tests.items()]
        done, not_done = concurrent.futures.wait(futures)
        response = {future.name: future.result() for future in done}

        for key in ('bad-request', 'search-url', 'search-body', 'show', 'timeline', 'get-ok'):
            r = response[key]
            # Ignore timeouts. These happen quite often
            if r.status_code == CLIENT_TIMEOUT:
                continue
            if key == 'bad-request':
                self.assertEqual(r.status_code, BAD_REQUEST)
                continue
            self.assertEqual(r.status_code, OK)
            result = r.json()
            if key in ('search-url', 'search-body'):
                # Even if we ask for 2 tweets, Twitter may return less
                self.assertLessEqual(len(result['statuses']), 2)
            elif key == 'show':
                self.assertEqual(result['screen_name'].lower(), 'gramener')
            elif key == 'timeline':
                self.assertEqual(len(result), 2)
                self.assertEqual(result[0]['user']['screen_name'].lower(), 'gramener')

        self.assertIn(response['get-redirect'].status_code, {OK, CLIENT_TIMEOUT})
        self.assertIn(response['post-fail'].status_code, {METHOD_NOT_ALLOWED, CLIENT_TIMEOUT})

def test_stream_blend(self):
        proc = gramex.cache.Subprocess(
            self.args1, stream_stdout='list_out', stream_stderr='list_out', buffer_size='line')
        [wait(future) for future in proc.wait_for_exit()]
        eq_(set(proc.list_out), {self.msg(s) for s in ('OUT:0', 'OUT:1', 'ERR:0', 'ERR:1')})

        proc = gramex.cache.Subprocess(
            self.args1, stream_stdout='queue_out', stream_stderr='queue_out', buffer_size='line')
        [wait(future) for future in proc.wait_for_exit()]
        items = set()
        for index in range(proc.queue_out.qsize()):
            items.add(proc.queue_out.get_nowait())
        eq_(items, {self.msg(s) for s in ('OUT:0', 'OUT:1', 'ERR:0', 'ERR:1')})

def test_stream_blend(self):
        proc = gramex.cache.Subprocess(
            self.args1, stream_stdout='list_out', stream_stderr='list_out', buffer_size='line')
        [wait(future) for future in proc.wait_for_exit()]
        eq_(set(proc.list_out), {self.msg(s) for s in ('OUT:0', 'OUT:1', 'ERR:0', 'ERR:1')})

        proc = gramex.cache.Subprocess(
            self.args1, stream_stdout='queue_out', stream_stderr='queue_out', buffer_size='line')
        [wait(future) for future in proc.wait_for_exit()]
        items = set()
        for index in range(proc.queue_out.qsize()):
            items.add(proc.queue_out.get_nowait())
        eq_(items, {self.msg(s) for s in ('OUT:0', 'OUT:1', 'ERR:0', 'ERR:1')})

def test_stream_none(self):
        proc = gramex.cache.Subprocess(self.args)
        stdout, stderr = [wait(future) for future in proc.wait_for_exit()]
        eq_(stdout, self.msg('OUT:0'))
        eq_(stderr, self.msg('ERR:0'))

        proc = gramex.cache.Subprocess(self.args1)
        stdout, stderr = [wait(future) for future in proc.wait_for_exit()]
        eq_(stdout, self.msg('OUT:0') + self.msg('OUT:1'))
        eq_(stderr, self.msg('ERR:0') + self.msg('ERR:1'))

def test_stream_list(self):
        proc = gramex.cache.Subprocess(
            self.args, stream_stdout='list_out', stream_stderr='list_err', buffer_size='line')
        [wait(future) for future in proc.wait_for_exit()]
        eq_(proc.list_out, [self.msg('OUT:0')])
        eq_(proc.list_err, [self.msg('ERR:0')])

        proc = gramex.cache.Subprocess(
            self.args1, stream_stdout='list_out', stream_stderr='list_err', buffer_size='line')
        wait_till(lambda: len(proc.list_out) > 0)
        wait_till(lambda: len(proc.list_err) > 0)
        eq_(proc.list_out, [self.msg('OUT:0')])
        eq_(proc.list_err, [self.msg('ERR:0')])
        wait_till(lambda: len(proc.list_out) > 1)
        wait_till(lambda: len(proc.list_err) > 1)
        eq_(proc.list_out, [self.msg('OUT:0'), self.msg('OUT:1')])
        eq_(proc.list_err, [self.msg('ERR:0'), self.msg('ERR:1')])
        [wait(future) for future in proc.wait_for_exit()]

def test_stream_none(self):
        proc = gramex.cache.Subprocess(self.args)
        stdout, stderr = [wait(future) for future in proc.wait_for_exit()]
        eq_(stdout, self.msg('OUT:0'))
        eq_(stderr, self.msg('ERR:0'))

        proc = gramex.cache.Subprocess(self.args1)
        stdout, stderr = [wait(future) for future in proc.wait_for_exit()]
        eq_(stdout, self.msg('OUT:0') + self.msg('OUT:1'))
        eq_(stderr, self.msg('ERR:0') + self.msg('ERR:1'))

def test_stream_queue(self):
        proc = gramex.cache.Subprocess(
            self.args, stream_stdout='queue_out', stream_stderr='queue_err')
        [wait(future) for future in proc.wait_for_exit()]
        eq_(proc.queue_out.get(), self.msg('OUT:0'))
        eq_(proc.queue_err.get(), self.msg('ERR:0'))

        proc = gramex.cache.Subprocess(
            self.args1, stream_stdout='queue_out', stream_stderr='queue_err', buffer_size='line')
        eq_(proc.queue_out.get(), self.msg('OUT:0'))
        eq_(proc.queue_err.get(), self.msg('ERR:0'))
        eq_(proc.queue_out.get(), self.msg('OUT:1'))
        eq_(proc.queue_err.get(), self.msg('ERR:1'))
        [wait(future) for future in proc.wait_for_exit()]
        eq_(proc.queue_out.qsize(), 0)
        eq_(proc.queue_err.qsize(), 0)