Task queue, Vagrant changes

Author: mc-slava

Vagrantfile

@@ -1,5 +1,6 @@
 VAGRANT_VERSION = "2"
 VM_BOX = "bento/ubuntu-16.04"
+USE_GIT_VERSION = false
 
 hosts = [
     {name: "microblog", ip: "192.168.33.10"}
@@ -16,7 +17,12 @@ Vagrant.configure(VAGRANT_VERSION) do |config|
 
     hosts.each_with_index do |elem, index|
         config.vm.define elem[:name] do |machine|
+            if USE_GIT_VERSION
             machine.vm.synced_folder ".", "/vagrant", disabled: true
+            else
+            machine.vm.synced_folder ".", "/opt/microblog", type: "rsync",
+                rsync__exclude: File.readlines('.gitignore').each(&:chomp!)
+            end
             machine.vm.synced_folder "./vagrant", "/automated"
             machine.vm.network :private_network, ip: elem[:ip]
             machine.vm.hostname = elem[:name]

app/__init__.py

@@ -11,6 +11,8 @@
 from flask_babel import Babel, lazy_gettext as _l
 from elasticsearch import Elasticsearch
 from config import Config
+from redis import Redis
+import rq
 
 db = SQLAlchemy()
 migrate = Migrate()
@@ -37,6 +39,9 @@ def create_app(config_class=Config):
     app.elasticsearch = Elasticsearch([app.config['ELASTICSEARCH_URL']]) \
             if app.config['ELASTICSEARCH_URL'] else None
 
+    app.redis = Redis.from_url(app.config['REDIS_URL'])
+    app.task_queue = rq.Queue('microblog-tasks', connection=app.redis)
+
     if app.config['ERROR_HANDLER'] and app.config['ERROR_HANDLER'] == "simple":
         from app.errors import simple as errors_handler
         errors_handler.set_error_handler(app)

app/cli.py

@@ -1,5 +1,5 @@
 from app import db
-from app.models import User, Post, Message, Notification
+from app.models import User, Post, Message, Notification, Task
 from datetime import datetime, timedelta
 import click
 import os
@@ -83,4 +83,5 @@ def show_all_attrs(value):
 
     @app.shell_context_processor
     def make_shell_context():
-        return {'db': db, 'User': User, 'Post': Post, 'Message': Message, 'Notification': Notification}
+        return {'db': db, 'User': User, 'Post': Post, 'Message': Message,
+                    'Notification': Notification, 'Task': Task}

app/email.py

@@ -7,8 +7,14 @@ def send_async_email(app, msg):
     with app.app_context():
         mail.send(msg)
 
-def send_email(subject, sender, recipients, text_body, html_body):
+def send_email(subject, sender, recipients, text_body, html_body, attachments=None, sync=False):
     msg = Message(subject, sender=sender, recipients=recipients)
     msg.body = text_body
     msg.html = html_body
-    Thread(target=send_async_email, args=(current_app._get_current_object(), msg)).start()
+    if attachments:
+        for attachment in attachments:
+            msg.attach(*attachment)
+    if sync:
+        mail.send(msg)
+    else:
+        Thread(target=send_async_email, args=(current_app._get_current_object(), msg)).start()

app/main/routes.py

@@ -18,6 +18,16 @@ def before_request():
         g.search_form = SearchForm() if current_app.elasticsearch else None
     g.locale = str(get_locale())
 
+@bp.route("/export_posts")
+@login_required
+def export_posts():
+    if current_user.get_task_in_progress('export_posts'):
+        flash(_("An export task is in progress already"))
+    else:
+        current_user.launch_task('export_posts', _("Exporting posts..."))
+        db.session.commit()
+    return redirect(url_for('main.user', username=current_user.username))
+
 @bp.route("/notifications")
 @login_required
 def notifications():

app/models.py

@@ -7,6 +7,8 @@
 from hashlib import md5
 import jwt
 import json
+import redis
+import rq
 from app.search import add_to_index, query_index, remove_from_index
 
 followers = db.Table('followers',
@@ -78,10 +80,23 @@ class User(SearchableMixin, UserMixin, db.Model):
                         backref='recipient', lazy='dynamic')
     last_message_read_time = db.Column(db.DateTime)
     notifications = db.relationship('Notification', backref='user', lazy='dynamic')
+    tasks = db.relationship('Task', backref='user', lazy='dynamic')
 
     def __repr__(self):
         return '<User {}>'.format(self.username)
 
+    def launch_task(self, name, description, *args, **kwargs):
+        rq_job = current_app.task_queue.enqueue('app.tasks.' + name, self.id, *args, **kwargs)
+        task = Task(id=rq_job.get_id(), name=name, description=description, user=self)
+        db.session.add(task)
+        return task
+
+    def get_tasks_in_progress(self):
+        return Task.query.filter_by(user=self, complete=False).all()
+
+    def get_task_in_progress(self, name):
+        return Task.query.filter_by(name=name, user=self, complete=False).first()
+
     def add_notification(self, name, data):
         self.notifications.filter_by(name=name).delete()
         n = Notification(name=name, payload_json=json.dumps(data), user=self)
@@ -182,6 +197,24 @@ class Notification(db.Model):
     def get_data(self):
         return json.loads(str(self.payload_json))
 
+class Task(db.Model):
+    id = db.Column(db.String(36), primary_key=True)
+    name = db.Column(db.String(128), index=True)
+    description = db.Column(db.String(128))
+    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
+    complete = db.Column(db.Boolean, default=False)
+
+    def get_rq_job(self):
+        rq_job = None
+        try:
+            rq_job = rq.job.Job.fetch(self.id, connection=current_app.redis)
+        except expression as identifier:
+            rq_job = None
+        return rq_job
+
+    def get_progress(self):
+        job = get_rq_job()
+        return job.meta.get('progress', 0) if job is not None else 100
 #
 
 db.event.listen(db.session, 'before_commit', Post.before_commit)

app/tasks.py

@@ -0,0 +1,57 @@
+import time
+import sys
+import json
+from rq import get_current_job
+from app import create_app, db
+from app.models import Task, User, Post
+from app.email import send_email
+from flask import render_template
+
+app = create_app()
+app.app_context().push()
+
+def example(seconds):
+    job = get_current_job()
+    print('Starting task')
+    for i in range(seconds):
+        job.meta['progress'] = 100.0 * i / seconds
+        job.save_meta()
+        print(i)
+        time.sleep(1)
+    job.meta['progress'] = 100
+    job.save_meta()
+    print('Task completed')
+
+def _set_task_progress(progress):
+    job = get_current_job()
+    if job:
+        job.meta['progress'] = progress
+        job.save_meta()
+        task = Task.query.get(job.get_id())
+        task.user.add_notification('task_progress', {'task_id': job.get_id(), 'progress': progress})
+        if progress >= 100:
+            task.complete = True
+        db.session.commit()
+
+def export_posts(user_id):
+    try:
+        user = User.query.get(user_id)
+        _set_task_progress(0)
+        data = []
+        i = 0
+        total_posts = user.posts.count()
+        for post in user.posts.order_by(Post.timestamp.asc()):
+            data.append({'body': post.body, 'timestamp': post.timestamp.isoformat() + 'Z'})
+            time.sleep(5)
+            i += 1
+            _set_task_progress(100 * i // total_posts)
+        send_email('[Microblog] Your blog posts',
+                sender=app.config['ADMINS'][0],
+                recipients=[user.email],
+                text_body=render_template('email/export_posts.txt', user=user),
+                html_body=render_template('email/export_posts.html', user=user),
+                attachments=[('posts.json', 'application/json', json.dumps({'posts': data}, indent=4))],
+                sync=True)
+    except:
+        _set_task_progress(100)
+        app.logger.error('Unhandled exception', exc_info=sys.exc_info())

app/templates/email/export_posts.html

@@ -0,0 +1,4 @@
+<p>Dear {{ user.username }},</p>
+<p>Please find attached the archive of your posts that you requested.</p>
+<p>Sincerely,</p>
+<p>The Microblog Team</p>

app/templates/email/export_posts.txt

@@ -0,0 +1,7 @@
+Dear {{ user.username }},
+
+Please find attached the archive of your posts that you requested.
+
+Sincerely,
+
+The Microblog Team

app/templates/main/user.html

@@ -27,6 +27,13 @@ <h1>{{ _("User: %(username)s", username=user.username) }}</h1>
                     <span>
                         <a href="{{ url_for('main.edit_profile') }}">{{ _("Edit profile") }}</a>
                     </span>
+                        {% if not current_user.get_task_in_progress('export_posts') %}
+                            <span>
+                                <a href="{{ url_for('main.export_posts') }}">
+                                    {{ _("Export posts") }}
+                                </a>
+                            </span>
+                        {% endif %}
                     {% else %}
                         {% if not current_user.is_following(user) %}
                         <span>

app/templates/msg/messages.html

@@ -4,7 +4,6 @@
     <div class="row">
         <div class="col-md-8 col-md-offset-2">
             <h1>{{ _("Messages") }}</h1>
-            {% if messages %}
             {% for post in messages %}
                 {% include 'main/_post.html' %}
             {% else %}

automated.sh

@@ -20,14 +20,18 @@ DEBIAN_FRONTEND=noninteractive apt-get -y install python3 python3-venv python3-d
 DB_ROOT=$(python3 -c "import uuid; print(uuid.uuid4().hex)")
 debconf-set-selections <<< "mysql-server mysql-server/root_password password ${DB_ROOT}"
 debconf-set-selections <<< "mysql-server mysql-server/root_password_again password ${DB_ROOT}"
-DEBIAN_FRONTEND=noninteractive apt-get -y install mysql-server elasticsearch postfix supervisor nginx
+DEBIAN_FRONTEND=noninteractive apt-get -y install mysql-server elasticsearch postfix supervisor nginx redis-server
 
 sed -i 's/#START_DAEMON/START_DAEMON/' /etc/default/elasticsearch
 systemctl restart elasticsearch
 
-mkdir -p /opt/microblog
-cd /opt/microblog
-git clone https://github.com/avoidik/microblog .
+if [[ ! -d "/opt/microblog" ]]; then
+  mkdir -p /opt/microblog
+  cd /opt/microblog
+  git clone https://github.com/avoidik/microblog .
+else
+  cd /opt/microblog
+fi
 pip3 install -U pip setuptools wheel
 python3 -m venv venv
 source venv/bin/activate
@@ -71,6 +75,8 @@ flask seed --no-destructive
 chown -R vagrant:vagrant /opt/microblog
 
 echo "export FLASK_APP=\"microblog.py\"" >> /home/vagrant/.profile
+echo "export LC_ALL=C.UTF-8" >> /home/vagrant/.profile
+echo "export LANG=C.UTF-8" >> /home/vagrant/.profile
 
 supervisorctl reload
 service nginx reload

config.py

@@ -21,3 +21,4 @@ class Config(object):
     ERROR_HANDLER = os.environ.get("ERROR_HANDLER") or None
     LOG_TO_STDOUT = os.environ.get("LOG_TO_STDOUT")
     SESSION_COOKIE_NAME = "microblog"
+    REDIS_URL = os.environ.get('REDIS_URL') or 'redis://'

migrations/versions/a93eeb179002_tasks.py

@@ -0,0 +1,38 @@
+"""tasks
+
+Revision ID: a93eeb179002
+Revises: c8af8ee4d608
+Create Date: 2018-01-27 23:31:36.434246
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'a93eeb179002'
+down_revision = 'c8af8ee4d608'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('task',
+    sa.Column('id', sa.String(length=36), nullable=False),
+    sa.Column('name', sa.String(length=128), nullable=True),
+    sa.Column('description', sa.String(length=128), nullable=True),
+    sa.Column('user_id', sa.Integer(), nullable=True),
+    sa.Column('complete', sa.Boolean(), nullable=True),
+    sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_task_name'), 'task', ['name'], unique=False)
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f('ix_task_name'), table_name='task')
+    op.drop_table('task')
+    # ### end Alembic commands ###

requirements.txt

@@ -14,4 +14,5 @@ python-dotenv
 elasticsearch
 gunicorn
 pymysql
-psycopg2
+psycopg2
+rq