本文整理汇总了Python中store.path函数的典型用法代码示例。如果您正苦于以下问题:Python path函数的具体用法?Python path怎么用?Python path使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了path函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: lookup
def lookup():
replies = []
for fn in os.listdir(g.loc):
if fn.startswith('reply-'):
try:
msg = crypto_util.decrypt(g.sid, g.codename,
file(store.path(g.sid, fn)).read()).decode("utf-8")
except UnicodeDecodeError:
app.logger.error("Could not decode reply %s" % fn)
else:
date = str(datetime.fromtimestamp(
os.stat(store.path(g.sid, fn)).st_mtime))
replies.append(dict(id=fn, date=date, msg=msg))
def async_genkey(sid, codename):
with app.app_context():
background.execute(lambda: crypto_util.genkeypair(sid, codename))
# Generate a keypair to encrypt replies from the journalist
# Only do this if the journalist has flagged the source as one
# that they would like to reply to. (Issue #140.)
if not crypto_util.getkey(g.sid) and g.source.flagged:
async_genkey(g.sid, g.codename)
return render_template('lookup.html', codename=g.codename, msgs=replies,
flagged=g.source.flagged, haskey=crypto_util.getkey(g.sid))
开发者ID:akiellor,项目名称:securedrop,代码行数:26,代码来源:source.py
示例2: lookup
def lookup():
msgs = []
flagged = False
for fn in os.listdir(g.loc):
if fn == '_FLAG':
flagged = True
continue
if fn.startswith('reply-'):
msgs.append(dict(
id=fn,
date=str(
datetime.fromtimestamp(
os.stat(store.path(g.sid, fn)).st_mtime)),
msg=crypto_util.decrypt(
g.sid, g.codename, file(store.path(g.sid, fn)).read())
))
if flagged:
session['flagged'] = True
def async_genkey(sid, codename):
with app.app_context():
background.execute(lambda: crypto_util.genkeypair(sid, codename))
# Generate a keypair to encrypt replies from the journalist
# Only do this if the journalist has flagged the source as one
# that they would like to reply to. (Issue #140.)
if not crypto_util.getkey(g.sid) and flagged:
async_genkey(g.sid, g.codename)
return render_template(
'lookup.html', codename=g.codename, msgs=msgs, flagged=flagged,
haskey=crypto_util.getkey(g.sid))
开发者ID:misjoinder,项目名称:cartodrop,代码行数:32,代码来源:source.py
示例3: create
def create():
sid = crypto_util.hash_codename(session['codename'])
if os.path.exists(store.path(sid)):
# if this happens, we're not using very secure crypto
app.logger.warning("Got a duplicate ID '%s'" % sid)
else:
os.mkdir(store.path(sid))
session['logged_in'] = True
session['flagged'] = False
return redirect(url_for('lookup'))
开发者ID:AllThing,项目名称:securedrop,代码行数:10,代码来源:source.py
示例4: create
def create():
sid = crypto_util.shash(session["codename"])
if os.path.exists(store.path(sid)):
# if this happens, we're not using very secure crypto
store.log("Got a duplicate ID '%s'" % sid)
else:
os.mkdir(store.path(sid))
session["logged_in"] = True
session["flagged"] = False
return redirect(url_for("lookup"))
开发者ID:ryanj,项目名称:securedrop,代码行数:10,代码来源:source.py
示例5: POST
def POST(self):
i = web.input('id', fh={}, msg=None, mid=None, action=None)
sid = crypto.shash(i.id)
if os.path.exists(store.path(sid)):
# if this happens, we're not using very secure crypto
store.log('Got a duplicate ID.')
else:
os.mkdir(store.path(sid))
return store_endpoint(i)
开发者ID:casey-bowman,项目名称:securedrop,代码行数:10,代码来源:source.py
示例6: POST
def POST(self):
iid = crypto.genrandomid()
if os.path.exists(store.path(crypto.shash(iid))):
# if this happens, we're not using very secure crypto
store.log('Got a duplicate ID.')
else:
os.mkdir(store.path(crypto.shash(iid)))
web.header('Cache-Control', 'no-cache, no-store, must-revalidate')
web.header('Pragma', 'no-cache')
web.header('Expires', '-1')
return render.generate(iid)
开发者ID:1mahesh,项目名称:deaddrop,代码行数:12,代码来源:source.py
示例7: get_docs
def get_docs(sid):
"""Get docs associated with source id `sid`, sorted by submission date"""
docs = []
for filename in os.listdir(store.path(sid)):
os_stat = os.stat(store.path(sid, filename))
docs.append(dict(
name=filename,
date=str(datetime.fromtimestamp(os_stat.st_mtime)),
size=os_stat.st_size,
))
# sort by date since ordering by filename is meaningless
docs.sort(key=lambda x: x['date'])
return docs
开发者ID:akiellor,项目名称:securedrop,代码行数:13,代码来源:journalist.py
示例8: get_docs
def get_docs(sid):
"""Get docs associated with source id `sid` sorted by submission date"""
docs = []
flagged = False
for filename in os.listdir(store.path(sid)):
if filename == "_FLAG":
flagged = True
continue
os_stat = os.stat(store.path(sid, filename))
docs.append(dict(name=filename, date=str(datetime.fromtimestamp(os_stat.st_mtime)), size=os_stat.st_size))
# sort by date since ordering by filename is meaningless
docs.sort(key=lambda x: x["date"])
return docs, flagged
开发者ID:ryanj,项目名称:securedrop,代码行数:13,代码来源:journalist.py
示例9: create
def create():
sid = crypto_util.hash_codename(session['codename'])
source = Source(sid, crypto_util.display_id())
db_session.add(source)
db_session.commit()
if os.path.exists(store.path(sid)):
# if this happens, we're not using very secure crypto
log.warning("Got a duplicate ID '%s'" % sid)
else:
os.mkdir(store.path(sid))
session['logged_in'] = True
return redirect(url_for('lookup'))
开发者ID:akiellor,项目名称:securedrop,代码行数:15,代码来源:source.py
示例10: GET
def GET(self, sid):
fns = os.listdir(store.path(sid))
docs = []
for f in fns:
docs.append(web.storage(
name=f,
date=str(datetime.datetime.fromtimestamp(os.stat(store.path(sid, f)).st_mtime))
))
docs.sort(lambda x,y: cmp(x.date, y.date))
haskey = bool(crypto.getkey(sid))
web.header('Cache-Control', 'no-cache, no-store, must-revalidate')
web.header('Pragma', 'no-cache')
web.header('Expires', '-1')
return render.col(docs, sid, haskey, codename=crypto.displayid(sid))
开发者ID:casey-bowman,项目名称:securedrop,代码行数:16,代码来源:journalist.py
示例11: GET
def GET(self):
dirs = os.listdir(config.STORE_DIR)
cols = []
for d in dirs:
if not os.listdir(store.path(d)): continue
cols.append(web.storage(name=d, codename=crypto.displayid(d), date=
str(datetime.datetime.fromtimestamp(
os.stat(store.path(d)).st_mtime
)).split('.')[0]
))
cols.sort(lambda x,y: cmp(x.date, y.date), reverse=True)
web.header('Cache-Control', 'no-cache, no-store, must-revalidate')
web.header('Pragma', 'no-cache')
web.header('Expires', '-1')
return render.index(cols)
开发者ID:1mahesh,项目名称:deaddrop,代码行数:16,代码来源:journalist.py
示例12: bulk_download
def bulk_download(sid, docs_selected):
source = get_source(sid)
filenames = [store.path(sid, doc['name']) for doc in docs_selected]
zip = store.get_bulk_archive(filenames)
return send_file(zip.name, mimetype="application/zip",
attachment_filename=source.journalist_designation + ".zip",
as_attachment=True)
开发者ID:akiellor,项目名称:securedrop,代码行数:7,代码来源:journalist.py
示例13: reply
def reply(journalist, source, num_replies):
"""Generates and submits *num_replies* replies to *source*
from *journalist*. Returns reply objects as a list.
:param db.Journalist journalist: The journalist to write the
reply from.
:param db.Source source: The source to send the reply to.
:param int num_replies: Number of random-data replies to make.
:returns: A list of the :class:`db.Reply`s submitted.
"""
assert num_replies >= 1
replies = []
for _ in range(num_replies):
source.interaction_count += 1
fname = "{}-{}-reply.gpg".format(source.interaction_count,
source.journalist_filename)
crypto_util.encrypt(str(os.urandom(1)),
[
crypto_util.getkey(source.filesystem_id),
config.JOURNALIST_KEY
],
store.path(source.filesystem_id, fname))
reply = db.Reply(journalist, source, fname)
replies.append(reply)
db.db_session.add(reply)
db.db_session.commit()
return replies
开发者ID:freedomofpress,项目名称:securedrop,代码行数:31,代码来源:db_helper.py
示例14: test_delete_collection
def test_delete_collection(self):
"""Test the "delete collection" button on each collection page"""
# first, add a source
self.source_app.get('/generate')
self.source_app.post('/create')
self.source_app.post('/submit', data=dict(
msg="This is a test.",
fh=(StringIO(''), ''),
), follow_redirects=True)
rv = self.journalist_app.get('/')
# navigate to the collection page
soup = BeautifulSoup(rv.data)
first_col_url = soup.select('ul#cols > li a')[0]['href']
rv = self.journalist_app.get(first_col_url)
self.assertEqual(rv.status_code, 200)
# find the delete form and extract the post parameters
soup = BeautifulSoup(rv.data)
delete_form_inputs = soup.select('form#delete_collection')[0]('input')
sid = delete_form_inputs[1]['value']
col_name = delete_form_inputs[2]['value']
rv = self.journalist_app.post('/col/delete/' + sid,
follow_redirects=True)
self.assertEquals(rv.status_code, 200)
self.assertIn(escape("%s's collection deleted" % (col_name,)), rv.data)
self.assertIn("No documents have been submitted!", rv.data)
# Make sure the collection is deleted from the filesystem
self._wait_for(
lambda: self.assertFalse(os.path.exists(store.path(sid)))
)
开发者ID:DarkDare,项目名称:securedrop,代码行数:33,代码来源:test_unit_integration.py
示例15: helper_filenames_delete
def helper_filenames_delete(self, soup, i):
filesystem_id = soup.select('input[name="filesystem_id"]')[0]['value']
checkbox_values = [
soup.select('input[name="doc_names_selected"]')[i]['value']]
# delete
resp = self.journalist_app.post('/bulk', data=dict(
filesystem_id=filesystem_id,
action='confirm_delete',
doc_names_selected=checkbox_values
), follow_redirects=True)
self.assertEqual(resp.status_code, 200)
self.assertIn(
"The following file has been selected for"
" <strong>permanent deletion</strong>",
resp.data)
# confirm delete
resp = self.journalist_app.post('/bulk', data=dict(
filesystem_id=filesystem_id,
action='delete',
doc_names_selected=checkbox_values
), follow_redirects=True)
self.assertEqual(resp.status_code, 200)
self.assertIn("Submission deleted.", resp.data)
# Make sure the files were deleted from the filesystem
utils.async.wait_for_assertion(lambda: self.assertFalse(
any([os.path.exists(store.path(filesystem_id, doc_name))
for doc_name in checkbox_values])))
开发者ID:freedomofpress,项目名称:securedrop,代码行数:30,代码来源:test_integration.py
示例16: test_delete_collections
def test_delete_collections(self, async_genkey):
"""Test the "delete selected" checkboxes on the index page that can be
used to delete multiple collections"""
# first, add some sources
num_sources = 2
for i in range(num_sources):
self.source_app.get('/generate')
self.source_app.post('/create')
self.source_app.post('/submit', data=dict(
msg="This is a test " + str(i) + ".",
fh=(StringIO(''), ''),
), follow_redirects=True)
self.source_app.get('/logout')
resp = self.journalist_app.get('/')
# get all the checkbox values
soup = BeautifulSoup(resp.data, 'html.parser')
checkbox_values = [checkbox['value'] for checkbox in
soup.select('input[name="cols_selected"]')]
resp = self.journalist_app.post('/col/process', data=dict(
action='delete',
cols_selected=checkbox_values
), follow_redirects=True)
self.assertEqual(resp.status_code, 200)
self.assertIn("%s collections deleted" % (num_sources,), resp.data)
self.assertTrue(async_genkey.called)
# Make sure the collections are deleted from the filesystem
utils.async.wait_for_assertion(lambda: self.assertFalse(
any([os.path.exists(store.path(filesystem_id))
for filesystem_id in checkbox_values])))
开发者ID:freedomofpress,项目名称:securedrop,代码行数:32,代码来源:test_integration.py
示例17: bulk_download
def bulk_download(sid, docs_selected):
filenames = [store.path(sid, doc["name"]) for doc in docs_selected]
zip = store.get_bulk_archive(filenames)
return send_file(
zip.name,
mimetype="application/zip",
attachment_filename=db.display_id(sid, db.sqlalchemy_handle()) + ".zip",
as_attachment=True,
)
开发者ID:ryanj,项目名称:securedrop,代码行数:9,代码来源:journalist.py
示例18: reply
def reply():
sid, msg_candidate = request.form["sid"], request.form["msg"]
try:
msg = msg_candidate.decode()
except (UnicodeDecodeError, UnicodeEncodeError):
flash("You have entered text that we could not parse. Please try again.", "notification")
return render_template("col.html", sid=sid, codename=db.display_id(sid, db.sqlalchemy_handle()))
crypto_util.encrypt(crypto_util.getkey(sid), msg, output=store.path(sid, "reply-%s.gpg" % uuid.uuid4()))
return render_template("reply.html", sid=sid, codename=db.display_id(sid, db.sqlalchemy_handle()))
开发者ID:ryanj,项目名称:securedrop,代码行数:9,代码来源:journalist.py
示例19: bulk_delete
def bulk_delete(sid, items_selected):
for item in items_selected:
item_path = store.path(sid, item.filename)
worker.enqueue(store.secure_unlink, item_path)
db_session.delete(item)
db_session.commit()
flash("Submission{} deleted.".format("s" if len(items_selected) > 1 else ""), "notification")
return redirect(url_for("col", sid=sid))
开发者ID:fowlslegs,项目名称:securedrop,代码行数:9,代码来源:journalist.py
示例20: doc
def doc(sid, fn):
if ".." in fn or fn.startswith("/"):
abort(404)
try:
Submission.query.filter(Submission.filename == fn).one().downloaded = True
except NoResultFound as e:
app.logger.error("Could not mark " + fn + " as downloaded: %s" % (e,))
db_session.commit()
return send_file(store.path(sid, fn), mimetype="application/pgp-encrypted")
开发者ID:fowlslegs,项目名称:securedrop,代码行数:9,代码来源:journalist.py
注:本文中的store.path函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论