From 85df4b3dc87003ae738f2676a99b88a83f3ac05a Mon Sep 17 00:00:00 2001 From: "B. Stack" Date: Sun, 13 Feb 2022 21:47:21 -0500 Subject: add expiry, and wsgi usage to support that The flask dev server seems incapable of using the @timer decorator, so we need the whole wsgi implementation, for which I've added a script. Arbitrary expiration options are available in the config file. --- pastebin.py | 83 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 78 insertions(+), 5 deletions(-) mode change 100644 => 100755 pastebin.py (limited to 'pastebin.py') diff --git a/pastebin.py b/pastebin.py old mode 100644 new mode 100755 index 3851d01..d06dc6e --- a/pastebin.py +++ b/pastebin.py @@ -1,7 +1,14 @@ -from datetime import datetime +from datetime import datetime, timedelta from itsdangerous import Signer from flask import (Flask, request, url_for, redirect, g, render_template, session, abort) from flask_sqlalchemy import SQLAlchemy +from pytimeparse.timeparse import timeparse # python3-pytimeparse +# uwsgidecorators load will fail when using initdb.py but is also not necessary +try: + from uwsgidecorators import timer # python3-uwsgidecorators +except: + pass +import time ## ripped from https://stackoverflow.com/questions/183042/how-can-i-use-uuids-in-sqlalchemy/812363#812363 from sqlalchemy import types @@ -12,6 +19,7 @@ class UUID(types.TypeDecorator): impl = MSBinary def __init__(self): self.impl.length = 16 + self.cache_ok = False # to shut up some useless warning types.TypeDecorator.__init__(self,length=self.impl.length) def process_bind_param(self,value,dialect=None): if value and isinstance(value,uuid.UUID): @@ -66,17 +74,27 @@ class Paste(db.Model): code = db.Column(db.Text) title = db.Column(db.Text) pub_date = db.Column(db.DateTime) + exp_date = db.Column(db.DateTime) user_id = db.Column(db.Integer, db.ForeignKey('user.id')) is_private = db.Column(db.Boolean) parent_id = db.Column(UUID(), db.ForeignKey('paste.id')) parent = db.relationship('Paste', lazy=True, backref='children', uselist=False, remote_side=[id]) - def __init__(self, user, code, title, parent=None, is_private=False): + def __init__(self, user, code, title, relative_expiration_seconds, parent=None, is_private=False): self.user = user self.code = code self.title = title self.is_private = is_private - self.pub_date = datetime.utcnow() + u = datetime.utcnow() + try: + # this will fail on POSTed value of "never" for exp + b = timedelta(seconds=relative_expiration_seconds) + except: + # so timedelta() will return 0 seconds, which makes the exp_date = pub_date, which is treated + # by the cleanup logic and jinja2 templates as never-expires. + b = timedelta() + self.pub_date = u + self.exp_date = u + b self.parent = parent class User(db.Model): @@ -99,13 +117,62 @@ def new_paste(): title = "Untitled paste" if request.form['pastetitle'] and request.form['pastetitle'] != "Enter title here": title = request.form['pastetitle'] - paste = Paste(g.user, request.form['code'], title, parent=parent, is_private=is_private) + relative_expiration_seconds = 0 + exp = 0 # start with an empty value + if 'exp' in request.form and request.form['exp']: + exp_opt = request.form['exp'] + if exp_opt not in app.config['EXPIRATION_OPTIONS']: + try: + exp = timeparse(f"+ {app.config['EXPIRATION_OPTIONS'][0]}") + except: + exp = 60 * 60 # failsafe, 1 hour + print(f"WARNING: requested expiration \"{exp_opt}\" is not in the list of options {app.config['EXPIRATION_OPTIONS']}, so will use {exp}") + else: + try: + exp = timeparse(f"+ {exp_opt}") + except: + exp = 0 + paste = Paste( + g.user, + request.form['code'], + title, + exp, + parent=parent, + is_private=is_private + ) db.session.add(paste) db.session.commit() sign = get_signed(paste.id, salt=app.config['SALT']) \ if is_private else None return redirect(url_for('show_paste', paste_id=paste.id, s=sign)) - return render_template('new_paste.html', parent=parent) + try: + exp_opts = app.config['EXPIRATION_OPTIONS'] + except: + exp_opts = None + return render_template('new_paste.html', parent=parent, exp_opts = exp_opts) + +# This @timer is from the uwsgidecorators +try: + @timer(app.config['LOOP_DELAY']) + def cleanup_expired_pastes(num): + # num is useless. + """ + Every LOOP_DELAY seconds, find any entries that have expired and then delete them. + """ + all1 = Paste.query.all() + need_commit = False + for p in all1: + # the exp_date != pub_date is very important, because anything with "never" expires + # is stored in the db as exp_date = pub_date + if p.exp_date and p.exp_date != p.pub_date and p.exp_date <= datetime.utcnow(): + print(f"INFO: deleting paste \"{p.title}\" with expiration {p.exp_date}.") + Paste.query.filter(Paste.id == p.id).delete() + need_commit = True + # only run the commit once! + if need_commit: + db.session.commit() +except: + pass @app.route('//') @app.route('/') @@ -188,6 +255,8 @@ def get_all_pastes(): p2 = { "id": p1.id, "title": p1.title, + "pub_date": p1.pub_date, + "exp_date": p1.exp_date, "private": private, "user_id": p1.user_id, "is_private": p1.is_private, @@ -203,3 +272,7 @@ def get_all_pastes(): def admin(): all_pastes = get_all_pastes() return render_template('admin.html', pastes = all_pastes) + +if __name__ == "__main__": + manager.add_command('runserver', Server(host=app.config["APP_HOST"], port=app.config["APP_PORT"])) + app.run() -- cgit