Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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 setupClass(cls):
gramex.cache.open(os.path.join(tests_dir, 'gramex.yaml'), 'config')
host = variables['REDIS_SERVER']
import redis
cls.redis = redis.StrictRedis(host=host, decode_responses=True, encoding='utf-8')
try:
# Re-initialize the database by clearing it
cls.redis.flushdb()
except redis.exceptions.ConnectionError:
raise SkipTest('No redis server at %s' % host)
cls.plainstore = RedisStore(path=host, flush=None)
cls.store = RedisStore(path='%s:6379' % host, flush=None,
purge_keys=BaseMixin._purge_keys)
cls.store2 = RedisStore(path='%s:6379:0' % host, flush=None,
purge_keys=BaseMixin._purge_keys)
out = gramex.data.download(AttrDict([
('dummy', self.dummy),
('sales', self.sales)
]), format='html')
result = pd.read_html(io.BytesIO(out), encoding='utf-8')
afe(result[0], self.dummy, check_column_type=six.PY3)
afe(result[1], self.sales, check_column_type=six.PY3)
def test_template(self):
raise SkipTest('TODO')
class FilterColsMixin(object):
sales = gramex.cache.open(sales_file, 'xlsx')
census = gramex.cache.open(sales_file, 'xlsx', sheet_name='census')
def unique_of(self, data, cols):
return data.groupby(cols).size().reset_index().drop(0, 1)
def check_keys(self, result, keys):
eq_(set(result.keys()), set(keys))
def test_filtercols_frame(self):
# ?_c=District returns up to 100 distinct values of the District column like
# {'District': ['d1', 'd2', 'd3', ...]}
cols = ['District']
args = {'_c': cols}
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(100))
def test_memory_cache_size(self):
memcache = gramex.services.info.cache['memory']
old_keys = set(memcache.keys())
data = gramex.cache.open(os.path.join(folder, 'sales.xlsx'))
new_keys = set(memcache.keys()) - old_keys
eq_(len(new_keys), 1) # only 1 new key should have been added
value = memcache[list(new_keys)[0]]
ok_(gramex.cache.sizeof(value) > sys.getsizeof(data))
uicomponents_path=uicomponents_path.replace('\\', '/'),
bootstrap_path=bootstrap_path.replace('\\', '/'),
google_fonts=google_fonts,
)
handle.write(result)
# Run sass to generate the output
options = ['--output-style', 'compressed']
proc = gramex.cache.Subprocess(
['node', sass_path, scss_path, cache_path] + options)
out, err = yield proc.wait_for_exit()
if proc.proc.returncode:
app_log.error('node-sass error: %s', err)
raise RuntimeError('Compilation failure')
handler.set_header('Content-Type', 'text/css')
raise Return(gramex.cache.open(cache_path, 'bin', mode='rb'))
for name, info in config['color'].items():
variables[name] = handler.get_arg(name, info['default'])
for name, info in config['toggle'].items():
variables.update(info['options'][handler.get_arg(name, info['default'])])
for name, info in config['font'].items():
val = info['options'][handler.get_arg(name, info['default'])]
if val['stack'] is not None:
variables[name] = val['stack']
if val['google']:
handle.write('@import url("https://fonts.googleapis.com/css?family=%s");\n' %
val['google'])
for key, val in variables.items():
handle.write('$%s: %s;\n' % (key, val))
handle.write('@import "%s";\n' % bootstrap_path.replace('\\', '/'))
# Run sass to generate the output
proc = gramex.cache.Subprocess(['node', sass_path, scss_path, cache_path])
out, err = yield proc.wait_for_exit()
with io.open(cache_path, 'rb') as handle:
raise Return(handle.read())
def load_data(data_config, handler=None):
'''
Loads data using gramex cache.
'''
if not isinstance(data_config, (dict, AttrDict,)):
raise ValueError('Data argument must be a dict like object.')
data = {}
for key, conf in data_config.items():
if isinstance(conf, (dict, AttrDict,)):
if 'function' in conf:
data[key] = build_transform(conf, vars={'handler': None})(handler=handler)[0]
elif conf.get('ext') in {'yaml', 'yml', 'json'}:
data[key] = gramex.cache.open(conf.pop('url'), conf.pop('ext'), **dict(conf))
elif 'url' in conf:
data[key] = gramex.data.filter(conf.pop('url'), **dict(conf))
else:
data[key] = conf
return data
def last_login():
"""Get last login details."""
user_logs_path = os.path.join(
variables.GRAMEXDATA, 'authmodule', 'user.csv')
names = ['time', 'event', 'sid', 'user', 'ip', 'user-agent']
if os.path.exists(user_logs_path):
data = gramex.cache.open(
user_logs_path, 'csv', header=None, names=names)
dt = datetime.datetime.strptime(
data.tail(1)['time'].values[0], "%Y-%m-%d %H:%M:%SZ")
return dt
return ''
def suggestion(handler):
'''first 3 questions as suggestion'''
return cache.open(file_path).sample(3)['Question'].to_json(orient='values')