Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def delete(self):
if self.driver_name != 'sqlalchemy':
raise NotImplementedError('driver=%s is not supported yet.' % self.driver_name)
kwargs = {'_wheres': self.getq('where')}
if self.thread:
self.result = yield gramex.service.threadpool.submit(self._sqlalchemy_delete, **kwargs)
else:
self.result = self._sqlalchemy_delete(**kwargs)
self.renderdata()
def post(self):
if self.driver_name != 'sqlalchemy':
raise NotImplementedError('driver=%s is not supported yet.' % self.driver_name)
kwargs = {'_vals': self.getq('val', [])}
if self.thread:
self.result = yield gramex.service.threadpool.submit(self._sqlalchemy_post, **kwargs)
else:
self.result = self._sqlalchemy_post(**kwargs)
self.renderdata()
def post(self, *args, **kwargs):
if self.redirects:
self.save_redirect_page()
upload = yield gramex.service.threadpool.submit(self.uploader.addfiles, self)
delete = yield gramex.service.threadpool.submit(self.uploader.deletefiles, self)
self.set_header('Content-Type', 'application/json')
self.write(json.dumps({'upload': upload, 'delete': delete},
ensure_ascii=True, separators=(',', ':')))
if self.redirects:
self.redirect_next()
def sendmail(handler):
mailer = gramex.service.email[handler.get_argument('from')]
folder = os.path.dirname(os.path.abspath(__file__))
gramex.service.threadpool.submit(
mailer.mail,
to=handler.get_argument('to'),
subject=handler.get_argument('subject', 'Gramex guide test'),
html='<p>This is a <strong>test email</strong> from ' +
'<a href="https://learn.gramener.com/guide/email/mail">Gramex</a></p>' +
'<p><img src="cid:kitten"></p>',
body='This is a plain text email from Gramex for non-HTML browsers',
attachments=[os.path.join(folder, 'README.md')],
images={'kitten': os.path.join(folder, 'kitten.jpg')},
)
return 'Mail is being sent. <a href=".">Back to email app</a>'
def post(self, *args, **kwargs):
if self.redirects:
self.save_redirect_page()
upload = yield gramex.service.threadpool.submit(self.uploader.addfiles, self)
delete = yield gramex.service.threadpool.submit(self.uploader.deletefiles, self)
self.set_header('Content-Type', 'application/json')
self.write(json.dumps({'upload': upload, 'delete': delete},
ensure_ascii=True, separators=(',', ':')))
if self.redirects:
self.redirect_next()
def get(self):
limit = int(self.getq('limit', [100])[0])
if isinstance(self.query, dict):
# Bind all queries and run them in parallel
args = {
key: self.getq(key, [''])[0]
for name, query in self.query.items()
for key, _bindparams in query._bindparams.items()
}
stmts = AttrDict([
(key, q.bindparams(**{k: v for k, v in args.items() if k in q._bindparams}))
for key, q in self.query.items()
])
if self.thread:
futures = AttrDict([
(key, gramex.service.threadpool.submit(self.run, stmt, limit))
for key, stmt in stmts.items()
])
self.result = AttrDict()
for key, future in futures.items():
self.result[key] = yield future
else:
self.result = AttrDict([
(key, self.run(stmt, limit)) for key, stmt in stmts.items()])
self.renderdatas()
else:
# Bind query and run it
args = {key: self.getq(key, [''])[0] for key in self.query._bindparams}
stmt = self.query.bindparams(**{k: v for k, v in args.items()
if k in self.query._bindparams})
if self.thread:
self.result = yield gramex.service.threadpool.submit(self.run, stmt, limit)
for key, stmt in stmts.items()
])
self.result = AttrDict()
for key, future in futures.items():
self.result[key] = yield future
else:
self.result = AttrDict([
(key, self.run(stmt, limit)) for key, stmt in stmts.items()])
self.renderdatas()
else:
# Bind query and run it
args = {key: self.getq(key, [''])[0] for key in self.query._bindparams}
stmt = self.query.bindparams(**{k: v for k, v in args.items()
if k in self.query._bindparams})
if self.thread:
self.result = yield gramex.service.threadpool.submit(self.run, stmt, limit)
else:
self.result = self.run(stmt, limit)
self.renderdata()
def languagetool(handler, *args, **kwargs):
import gramex
merge(kwargs, _languagetool['defaults'], mode='setdefault')
yield gramex.service.threadpool.submit(languagetool_download)
if not handler:
lang = kwargs.get('lang', 'en-us')
q = kwargs.get('q', '')
else:
lang = handler.get_argument('lang', 'en-us')
q = handler.get_argument('q', '')
result = yield languagetoolrequest(q, lang, **kwargs)
errors = json.loads(result.decode('utf8'))['matches']
if errors:
result = {
"errors": errors,
}
corrected = list(q)
d_offset = 0 # difference in the offset caused by the correction
for error in errors:
# only accept the first replacement for an error