Basic analysis and indexing

Author: eghuro

tsa/public/views.py

@@ -1,9 +1,12 @@
 # -*- coding: utf-8 -*-
 """Public section, including homepage and signup."""
-from collections import defaultdict
-from flask import abort, Blueprint, jsonify, render_template, request, url_for
+import logging
+import redis
 import rfc3987
-from tsa.tasks import hello, analyze
+from atenvironment import environment
+from collections import defaultdict
+from flask import abort, Blueprint, current_app, jsonify, render_template, request, url_for
+from tsa.tasks import hello, system_check, analyze, analyze_upload
 
 blueprint = Blueprint('public', __name__, static_folder='../static')
 
@@ -13,40 +16,78 @@ def home():
     """Landing page."""
     return render_template('public/landing.html')
 
-@blueprint.route('/api/1/test/base')
+
+@blueprint.route('/api/v1/test/base')
 def test_basic():
-    return ""
+    return "Hello world!"
 
-@blueprint.route('/api/1/test/job')
+
+@blueprint.route('/api/v1/test/job')
 def test_celery():
     r = hello.delay()
     return r.get()
 
-@blueprint.route('/api/1/analyze')
-def api_analyze():
+
+@blueprint.route('/api/v1/test/system')
+def test_system():
+    x = (system_check.s() | hello.si()).delay().get()
+    log = logging.getLogger(__name__)
+    log.info(f"System check result: {x!s}")
+    return str(x)
+
+
+@blueprint.route('/api/v1/analyze', methods=['GET'])
+def api_analyze_iri():
     iri = request.args.get('iri', None)
+    etl = bool(int(request.args.get('etl', 1)))
+
+    current_app.logger.info("ETL: " + str(etl))
+
     if rfc3987.match(iri):
-        task = analyze.delay(iri)
-        return "", 202, {'Location': url_for('public.check_status', task_id=task.id)}
+        return jsonify(analyze.delay(iri, etl).get())
     else:
         abort(400)
 
-@blueprint.route('/api/1/analyze/status/<task_id>')
-def check_status(task_id):
-    task = analyze.AsyncResult(task_id)
-    def default(value):
-        return { 'state': value.state, 'status': str(value.info) }
 
-    return jsonify(defaultdict(default,
-        PENDING={ 'state': task.state, 'status': 'Pending' },
-        SUCCESS={ 'state': task.state, 'status': 'Completed' },
-        FAILURE={ 'state': task.state, 'status': 'Failed' }
-    )[task.state])
+@blueprint.route('/api/v1/analyze', methods=['POST'])
+@environment('REDIS')
+def api_analyze_upload(redis_url):
+    etl = bool(int(request.args.get('etl', 1)))
+
+    def read_in_chunks(file_object, chunk_size=1024):
+        """Lazy function (generator) to read a file piece by piece.
+        Default chunk size: 1k."""
+        while True:
+            data = file_object.read(chunk_size)
+            if not data:
+                break
+            yield data
+
+    keys = []
+    mimes = []
+    r = redis.StrictRedis.from_url(redis_url, charset="utf-8", decode_responses=True)
+    for file in request.files:
+        key = str(uuid.uuid4())
+        keys.append(key)
+        mimes.append(file.mimetype)
+        for piece in read_in_chunks(file):
+            r.append(key, piece)
+        r.expire(key, 60)
+
+    g = group(analyze_upload.s(k, m, etl) for k, m in zip(keys, mimes))
+    return jsonify(g.apply_async().get())
 
-@blueprint.route('/api/1/query')
-def query():
+        
+@blueprint.route('/api/v1/query')
+@environment('REDIS')
+def index(redis_url):
+    r = redis.StrictRedis.from_url(redis_url, charset="utf-8", decode_responses=True)
     iri = request.args.get('iri', None)
+    current_app.logger.info("Querying for: " + iri)
     if rfc3987.match(iri):
-        return ""
+        if not r.exists(iri):
+            abort(404)
+        else:
+            return jsonify([str(x) for x in r.smembers(iri)])
     else:
         abort(400)

tsa/settings.py

@@ -19,15 +19,15 @@ class Config(object):
     CACHE_TYPE = 'redis'  # Can be "memcached", "redis", etc.
     SQLALCHEMY_TRACK_MODIFICATIONS = False
     WEBPACK_MANIFEST_PATH = 'webpack/manifest.json'
-    CELERY_BROKER_URL = 'redis://redis:6379/0'
-    CELERY_RESULT_BACKEND = 'redis://redis:6379/0'
+    CELERY_BROKER_URL = os.environ.get('REDIS', 'redis://redis:6379/0')
+    CELERY_RESULT_BACKEND = os.environ.get('REDIS', 'redis://redis:6379/0')
     REDIS_HOST = 'redis'
     REDIS_PORT = 6379
     REDIS_DB = 0
     CACHE_KEY_PREFIX = 'fcache'
     CACHE_REDIS_HOST = 'redis'
     CACHE_REDIS_PORT = '6379'
-    CACHE_REDIS_URL = 'redis://redis:6379/0'
+    CACHE_REDIS_URL = os.environ.get('REDIS', 'redis://redis:6379/0')
     SENTRY_CONFIG = {
         'dsn': 'https://9df1f926d1854fa4884d1f0ce9489a0b@sentry.io/1304923',
         'release': RELEASE,

tsa/tasks.py

@@ -1,10 +1,184 @@
 """Celery tasks invoked from the API endpoints."""
+import json
+import logging
+import rdflib
+import redis
+import requests
+from atenvironment import environment
+from rdflib import URIRef
+from urllib.parse import urlparse
+from urllib.error import URLError
+from tsa.analyzer import Analyzer
 from tsa.celery import celery
+from tsa.transformation import PipelineFactory
+
+@celery.task
+@environment('ETL', 'VIRTUOSO')
+def system_check(etl, virtuoso):
+    log = logging.getLogger(__name__)
+    log.info("System check started")
+    log.info(f"Testing LP-ETL, URL: {etl!s}")
+    requests.get(etl).raise_for_status()
+
+    virtuoso_url = f"{virtuoso!s}/sparql"
+    log.info(f"Testing virtuoso, URL: {virtuoso_url}")
+    requests.get(virtuoso_url).raise_for_status()
+
+    log.info("System check successful")
+
 
 @celery.task
 def hello():
     return "Hello world!"
 
+
+@celery.task
+def analyze(iri, etl=True):
+    log = logging.getLogger(__name__)
+    log.info(f"Analyzing {iri!s}")
+    if etl:
+        (transform.s(iri) | poll.s() | inspect.s()).apply_async()
+    else:
+        guess = rdflib.util.guess_format(iri)
+        if guess is None:
+            r = requests.head(iri)
+            r.raise_for_status()
+            guess = r.headers.get('content-type')
+        g = rdflib.ConjunctiveGraph()
+        log.info(f"Guessing format to be {guess!s}")
+        g.parse(iri, format=guess)
+        a = Analyzer()
+        index(g, iri)
+        return a.analyze(g)
+
+@environment('REDIS')
+def index(g, source_iri, redis_cfg):
+    r = redis.StrictRedis.from_url(redis_cfg)
+    pipe = r.pipeline()
+    exp = 60*60 #1H
+    for (s, p, o) in g:
+        s = str(s)
+        p = str(p)
+        o = str(o)
+        source_iri = str(source_iri)
+
+        pipe.sadd(s, source_iri, p, o)
+        pipe.sadd(p, source_iri, s, o)
+        pipe.sadd(o, source_iri, p, s)
+        pipe.sadd(source_iri, s, p, o)
+
+        pipe.expire(s, exp)
+        pipe.expire(p, exp)
+        pipe.expire(o, exp)
+        pipe.expire(source_iri, exp)
+    pipe.execute()
+
+@celery.task
+@environment('REDIS')
+def analyze_upload(key, mime, etl, redis_cfg):
+    log = logging.getLogger(__name__)
+    r = redis.StrictRedis.from_url(redis_cfg)
+    if r.strlen(key) < 1024 * 1024: #approx 1MB
+        g = rdflib.ConjunctiveGraph()
+        g.parse(data=r.get(key), format=mime)
+        a = Analyzer()
+        return a.analyze(g)
+    else:
+       log.warn(f"Not analyzing an upload as it's too big: {key!s}")
+       r.delete(key) 
+
+
+@celery.task
+def inspect(iri):
+    log = logging.getLogger(__name__)
+    g = rdflib.ConjunctiveGraph()
+    g.parse(iri)
+    a = Analyzer()
+    return a.analyze(g)
+
+
+@celery.task
+@environment('ETL', 'VIRTUOSO', 'DBA_PASSWORD')
+def transform(iri, etl, virtuoso, dbaPass):
+    log = logging.getLogger(__name__)
+    #create pipeline and call to start executions
+    # prepare JSON-LD pipeline
+
+    log.info(f"Prepare pipeline for {iri!s}")
+    pf = PipelineFactory()
+    p = urlparse(virtuoso)
+    pipeline = json.dumps(pf.createPipeline(iri, {'server': p.hostname, 'port': 1111, 'user': 'dba', 'password': dbaPass, 'iri': iri}))
+
+    log.info(f"Pipeline:\n{pipeline!s}")
+
+    # create the pipeline
+    r = requests.post(f"{etl!s}/resources/pipelines", files={'pipeline': pipeline})
+    r.raise_for_status()
+
+    g = rdflib.ConjunctiveGraph()
+    g.parse(data=r.text, format="trig")
+
+    pipeline = g.value(object=URIRef("http://linkedpipes.com/ontology/Pipeline"), predicate=rdflib.namespace.RDF.type)
+    log.info(f"Pipeline IRI: {pipeline!s}")
+
+    # POST /resources/executions
+    r = requests.post(f"{etl!s}/resources/executions?pipeline={pipeline}")
+    r.raise_for_status()
+    log.info(f"Execution trigger result:\n{r.json()!s}")
+    return f"{etl!s}/resources/executions/{r.json()['iri'].split('/')[-1]}"
+
+
+@celery.task(bind=True, retry_backoff=True, max_retries=None, default_retry_delay=30, time_limit=60*60)
+def poll(self, iri):
+    def after_return(self, status, retval, task_id, args, kwargs, einfo):
+        cleanup.apply_async()
+    self.after_return = after_return
+
+    log = logging.getLogger(__name__)
+    log.info(f"Polling {iri!s}")
+
+    r = requests.get(iri + "/overview")
+    content = r.text
+    log.info(content)
+    r.raise_for_status()
+
+    j = json.loads(content)
+    if j['status']['@id'] == "http://etl.linkedpipes.com/resources/status/failed":
+        log.error("Execution failed")
+
+        try:
+            r = requests.get(iri + "/logs")
+            r.raise_for_status()
+            log.error("ETL log:\n" + r.text)
+        except HTTPError as e:
+            raise EtlJobFailed(r) from e
+
+        raise EtlJobFailed(r)
+    elif not (j['status']['@id'] == "http://etl.linkedpipes.com/resources/status/finished"):
+        log.info("Execution is not finished yet")
+        self.retry()
+    else:
+        #get result uri
+        log.info(f"Final graph:\n{str(g)!s}")
+        result = ""
+        return result
+
+
 @celery.task
-def analyze(iri):
-    return ""
+@environment('ETL')
+def cleanup(iri, etl):
+    log = logging.getLogger(__name__)
+    log.info(f"Deleting {iri!s}")
+    
+    r = requests.delete(f"{etl!s}/pipelines?iri={iri!s}")
+    r.raise_for_status()
+
+    log.info(f"Pipeline {iri!s} deleted")
+
+
+class EtlError(Exception):
+    pass
+
+
+class EtlJobFailed(EtlError):
+    pass