Merge pull request #10 from imperial-genomics-facility/project_data

Merge Project data branch

.github/workflows/python-app.yml

@@ -5,21 +5,21 @@ name: Python application
 
 on:
   push:
-    branches: [ main ]
+    branches: [ main, project_data ]
   pull_request:
-    branches: [ main ]
+    branches: [ main, project_data ]
 
 jobs:
   build:
 
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@v2
-    - name: Set up Python 3.7
-      uses: actions/setup-python@v2
+    - uses: actions/checkout@v3
+    - name: Set up Python 3.8
+      uses: actions/setup-python@v4
       with:
-        python-version: "3.7"
+        python-version: "3.8"
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
@@ -33,4 +33,4 @@ jobs:
         flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
     - name: Test with pytest
       run: |
-        pytest
+        pytest --cov=app --log-level=ERROR tests

.gitignore

@@ -24,3 +24,11 @@ nginx.conf
 app/static/bclconvert_report_v0.02.html
 app/static/MultiQC_lane1.html
 app/static/MultiQC_lane1.html:Zone.Identifier
+app/static/*
+.coverage
+tests/1__init__.py
+aaa
+facebook.ico
+migration_docker_db.sh
+static/*
+celery_tmp/*

Dockerfile

@@ -1,5 +1,5 @@
-FROM python:3.7.12-slim
-LABEL version="v0.01"
+FROM python:3.8.16-slim
+LABEL version="v0.02"
 LABEL description="Docker image for running IGFPortal server"
 COPY requirements.txt /tmp/requirements.txt
 RUN apt-get -y update && \
@@ -13,4 +13,4 @@ RUN apt-get -y update && \
 USER nobody
 WORKDIR /tmp
 ENTRYPOINT ["bash","-c"]
-CMD ["flask", "run"]
+CMD ["flask", "run"]

app/__init__.py

@@ -1,9 +1,11 @@
+import os
 import logging
-
 from flask import Flask, request
 from flask_appbuilder import AppBuilder, SQLA
 from .index import CustomIndexView
 from celery import Celery
+from flask_caching import Cache
+from flask_migrate import Migrate
 
 """
  Logging configuration
@@ -15,6 +17,7 @@
 app = Flask(__name__)
 app.config.from_object("config")
 db = SQLA(app)
+migrate = Migrate(app, db)
 appbuilder = AppBuilder(app, db.session, indexview=CustomIndexView)
 
 
@@ -38,6 +41,28 @@ def set_sqlite_pragma(dbapi_connection, connection_record):
         result_backend=app.config['CELERY_RESULT_BACKEND'])
 celery.conf.update(app.config)
 
+## CACHING
+cache_config = {
+    "CACHE_TYPE": "RedisCache",
+    "CACHE_DEFAULT_TIMEOUT": 300,
+    "CACHE_REDIS_URL": app.config['CACHE_REDIS_URL']
+}
+
+test_cache_config = {
+    "CACHE_TYPE": "SimpleCache",
+    "CACHE_DEFAULT_TIMEOUT": 300,
+}
+
+env_name = os.environ.get('ENV_NAME')
+if 'TESTING' in app.config and \
+   app.config.get('TESTING') is not None and \
+   app.config.get('TESTING'):
+  app.config.from_mapping(test_cache_config)
+  cache = Cache(app)
+else:
+  app.config.from_mapping(cache_config)
+  cache = Cache(app)
+
 ## GDPR
 @app.context_processor
 def inject_template_scope():

app/admin_home_api.py

@@ -9,14 +9,16 @@
 from . import app, db, celery
 from .admin_home.admin_home_utils import parse_and_add_new_admin_view_data
 
+log = logging.getLogger(__name__)
+
 @celery.task(bind=True)
 def async_parse_and_add_new_admin_view_data(
     self, json_file: str) -> dict:
     try:
         parse_and_add_new_admin_view_data(json_file)
         return {"message": "success"}
     except Exception as e:
-        logging.error(
+        log.error(
             "Failed to run celery job, error: {0}".\
                 format(e))
 
@@ -43,9 +45,9 @@ def update_admin_view_data(self):
                     prefix='admin_view_',)
             with open(json_file, 'w') as fp:
                 json.dump(json_data, fp)
-            _ = \
+            msg = \
                 async_parse_and_add_new_admin_view_data.\
                     apply_async(args=[json_file])
             return self.response(200, message='loaded new data')
         except Exception as e:
-            logging.error(e)
+            log.error(e)

app/airflow/airflow_api_utils.py

@@ -1,13 +1,26 @@
 import json
 import requests
 from urllib.parse import urljoin
+from typing import Union
+
+def get_airflow_dag_id(airflow_conf_file: str, dag_tag: str) -> Union[str, None]:
+    try:
+        with open(airflow_conf_file, "r") as fp:
+            airflow_conf = json.load(fp)
+        dag_id = airflow_conf.get(dag_tag)
+        return dag_id
+    except Exception as e:
+        raise ValueError(
+            f"Failed to get dag id for tag {dag_tag} in config file {airflow_conf_file}, error: {e}")
+
 
 def post_to_airflow_api(
       airflow_conf_file: str,
       url_suffix: str,
       data: dict,
       headers: dict = {"Content-Type": "application/json"},
-      verify: bool = False):
+      verify: bool = False,
+      dry_run: bool = False):
     try:
         with open(airflow_conf_file, "r") as fp:
             airflow_conf = json.load(fp)
@@ -19,13 +32,14 @@ def post_to_airflow_api(
         url = \
             urljoin(airflow_conf['url'], url_suffix)
         res = \
-        requests.post(
-            url=url,
-            data=data,
-            headers=headers,
-            auth=(airflow_conf["username"], airflow_conf["password"]),
-            verify=verify)
-        if res.status_code != 200:
+            requests.post(
+                url=url,
+                data=data,
+                headers=headers,
+                auth=(airflow_conf["username"], airflow_conf["password"]),
+                verify=verify)
+        if res.status_code != 200 and \
+           not dry_run:
             raise ValueError(
                 f"Failed post request, got status: {res.status_code}")
         return res
@@ -36,16 +50,20 @@ def post_to_airflow_api(
 def trigger_airflow_pipeline(
       dag_id: str,
       conf_data: dict,
-      airflow_conf_file: str):
+      airflow_conf_file: str,
+      verify: bool = False,
+      dry_run: bool = False):
     try:
         url_suffix = \
             f'dags/{dag_id}/dagRuns'
         data = {"conf": conf_data}
         res = \
-        post_to_airflow_api(
-            airflow_conf_file=airflow_conf_file,
-            url_suffix=url_suffix,
-            data=data)
+            post_to_airflow_api(
+                airflow_conf_file=airflow_conf_file,
+                url_suffix=url_suffix,
+                data=data,
+                verify=verify,
+                dry_run=dry_run)
         return res
     except Exception as e:
         raise ValueError(

app/analysis_view.py

@@ -0,0 +1,121 @@
+import logging
+import os
+from app import db
+from .models import Analysis
+from .models import Pipeline_seed
+from .models import Pipeline
+from .airflow.airflow_api_utils import trigger_airflow_pipeline
+from flask_appbuilder import ModelView
+from flask import redirect, flash, url_for, send_file
+from flask_appbuilder.actions import action
+from flask_appbuilder.models.sqla.interface import SQLAInterface
+from . import celery
+
+log = logging.getLogger(__name__)
+
+def get_analysis_pipeline_seed_status(analysis_id: int) -> str:
+    try:
+        result = \
+            db.session.\
+                query(
+                    Analysis.analysis_name,
+                    Pipeline.pipeline_name,
+                    Pipeline_seed.status).\
+                join(Pipeline, Pipeline.pipeline_name==Analysis.analysis_type).\
+                join(Pipeline_seed, Pipeline_seed.seed_id==Analysis.analysis_id).\
+                filter(Pipeline_seed.pipeline_id==Pipeline.pipeline_id).\
+                filter(Pipeline_seed.seed_table=='analysis').\
+                filter(Pipeline_seed.status=='SEEDED').\
+                filter(Pipeline.pipeline_type=='AIRFLOW').\
+                filter(Analysis.analysis_id==analysis_id).\
+                one_or_none()
+        if result is None:
+            return 'INVALID'
+        else:
+            return 'VALID'
+    except Exception as e:
+        log.error(e)
+        raise ValueError(
+            f"Failed to get analysis pipeline seed status, error: {e}")
+
+
+@celery.task(bind=True)
+def async_submit_analysis_pipeline(self, id_list):
+    try:
+        results = list()
+        for analysis_id in id_list:
+            ## get dag id
+            dag_name = \
+                db.session.\
+                    query(Analysis.analysis_type).\
+                    filter(Analysis.analysis_id==analysis_id).\
+                    one_or_none()
+            if dag_name is not None:
+                dag_name = dag_name[0]
+            res = \
+                trigger_airflow_pipeline(
+                    dag_id=dag_name,
+                    conf_data={"analysis_id": analysis_id},
+                    airflow_conf_file=os.environ['AIRFLOW_CONF_FILE'])
+            results.append(res.status_code)
+        return dict(zip(id_list, results))
+    except Exception as e:
+        log.error(
+            f"Failed to run celery job, error: {e}")
+
+
+class AnalysisView(ModelView):
+    datamodel = \
+        SQLAInterface(Analysis)
+    list_columns = [
+        "analysis_name",
+        "analysis_type",
+        "project.project_igf_id"]
+    base_permissions = [
+        "can_list",
+        "can_show"]
+    base_order = ("analysis_id", "desc")
+
+    @action("trigger_analysis_pipeline", "Trigger analysis pipeline", confirmation="confirm pipeline run?", icon="fa-rocket")
+    def trigger_analysis_pipeline(self, item):
+        try:
+            id_list = list()
+            analysis_list = list()
+            if isinstance(item, list):
+                id_list = [i.analysis_id for i in item]
+                analysis_list = [i.analysis_name for i in item]
+            else:
+                id_list = [item.analysis_id]
+                analysis_list = [item.analysis_name]
+            analysis_dict = \
+                dict(zip(id_list, analysis_list))
+            invalid_id_list = list()
+            valid_id_list = list()
+            invalid_name_list = list()
+            valid_name_list = list()
+            for analysis_id in id_list:
+                status = \
+                    get_analysis_pipeline_seed_status(
+                        analysis_id=analysis_id)
+                if status == 'VALID':
+                    valid_id_list.\
+                        append(analysis_id)
+                    valid_name_list.\
+                        append(analysis_dict.get(analysis_id))
+                if status == 'INVALID':
+                    invalid_id_list.\
+                        append(analysis_id)
+                    invalid_name_list.\
+                        append(analysis_dict.get(analysis_id))
+            if len(valid_name_list) > 0:
+                _ = \
+                    async_submit_analysis_pipeline.\
+                        apply_async(args=[valid_id_list])
+                flash(f"Submitted jobs for {', '.join(valid_name_list)}", "info")
+            if len(invalid_name_list) > 0:
+                flash(f"Skipped old analysis {', '.join(invalid_name_list)}", "danger")
+            self.update_redirect()
+            return redirect(url_for('AnalysisView.list'))
+        except:
+            flash('Failed to submit analysis', 'danger')
+            return redirect(url_for('AnalysisView.list'))

app/apis.py

@@ -5,6 +5,7 @@
 from .raw_metadata_api import RawMetadataDataApi
 from .admin_home_api import AdminHomeApi
 from .raw_seqrun_api import RawSeqrunApi
+from .raw_analysis_api import RawAnalysisApi
 
 
 """
@@ -15,4 +16,5 @@
 appbuilder.add_api(MetadataLoadApi)
 appbuilder.add_api(RawMetadataDataApi)
 appbuilder.add_api(AdminHomeApi)
-appbuilder.add_api(RawSeqrunApi)
+appbuilder.add_api(RawSeqrunApi)
+appbuilder.add_api(RawAnalysisApi)

app/forms.py

@@ -2,7 +2,7 @@
 from flask_appbuilder.forms import DynamicForm
 from wtforms.fields import StringField,SubmitField,IntegerField,RadioField,DecimalField
 from wtforms.validators import DataRequired,InputRequired,NumberRange
-from wtforms.ext.sqlalchemy.fields import QuerySelectField
+from wtforms_sqlalchemy.fields import QuerySelectField
 from . import appbuilder, db
 from .models import IlluminaInteropData
 

app/home_view.py

@@ -2,6 +2,7 @@
 from flask_appbuilder.baseviews import BaseView, expose
 from flask_appbuilder.security.decorators import protect, has_access
 from . import db
+from app import cache
 from .models import AdminHomeData
 
 log = logging.getLogger(__name__)
@@ -41,11 +42,13 @@ class HomeView(BaseView):
 
     @expose('/user_home')
     @has_access
+    @cache.cached(timeout=600)
     def general(self):
         return self.render_template('user_index.html')
 
     @expose('/admin_home')
     @has_access
+    @cache.cached(timeout=600)
     def admin_home(self):
         try:
             (finished_seqrun, finished_analysis,