Lint

Author: eghuro

autoapp.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 """Create an application instance."""
 import logging
+
 from flask.helpers import get_debug_flag
 
 from tsa.app import create_app
@@ -10,8 +11,10 @@
 
 app = create_app(CONFIG)
 
+
 @app.before_first_request
 def setup_logging():
+    """Set up logging to STDOUT from INFO level up in production environment."""
     if not app.debug:
         # In production mode, add log handler to sys.stderr.
         app.logger.addHandler(logging.StreamHandler())

tsa/analyzer.py

@@ -1,8 +1,7 @@
-import json
-import uuid
 from collections import defaultdict
+
 from rdflib.namespace import RDF
-from tsa.extensions import redis
+
 
 class Analyzer(object):
 

tsa/app.py

@@ -3,7 +3,7 @@
 
 import logging
 
-from flask import Flask, render_template, g
+from flask import Flask, g, render_template
 
 from tsa import commands, public
 from tsa.extensions import bcrypt, cache, csrf_protect, db, debug_toolbar, migrate, sentry, webpack

tsa/cache.py

@@ -5,7 +5,7 @@
 import os
 from functools import wraps
 
-from flask import request, make_response
+from flask import make_response, request
 
 from tsa.extensions import cache
 

tsa/celery.py

@@ -2,7 +2,7 @@
 
 import celery
 import raven
-from raven.contrib.celery import register_signal, register_logger_signal
+from raven.contrib.celery import register_logger_signal, register_signal
 
 
 class Celery(celery.Celery):

tsa/public/views.py

@@ -1,12 +1,15 @@
 # -*- coding: utf-8 -*-
 """Public section, including homepage and signup."""
 import logging
+import uuid
+
 import redis
 import rfc3987
 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
+from celery import group
+from flask import Blueprint, abort, current_app, jsonify, render_template, request
+
+from tsa.tasks import analyze, analyze_upload, hello, system_check
 
 blueprint = Blueprint('public', __name__, static_folder='../static')
 
@@ -19,7 +22,7 @@ def home():
 
 @blueprint.route('/api/v1/test/base')
 def test_basic():
-    return "Hello world!"
+    return 'Hello world!'
 
 
 @blueprint.route('/api/v1/test/job')
@@ -32,7 +35,7 @@ def test_celery():
 def test_system():
     x = (system_check.s() | hello.si()).delay().get()
     log = logging.getLogger(__name__)
-    log.info(f"System check result: {x!s}")
+    log.info(f'System check result: {x!s}')
     return str(x)
 
 
@@ -41,7 +44,7 @@ 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))
+    current_app.logger.info(f'ETL:{etl!s}')
 
     if rfc3987.match(iri):
         return jsonify(analyze.delay(iri, etl).get())
@@ -56,7 +59,9 @@ def api_analyze_upload(redis_url):
 
     def read_in_chunks(file_object, chunk_size=1024):
         """Lazy function (generator) to read a file piece by piece.
-        Default chunk size: 1k."""
+
+        Default chunk size: 1k.
+        """
         while True:
             data = file_object.read(chunk_size)
             if not data:
@@ -65,7 +70,7 @@ def read_in_chunks(file_object, chunk_size=1024):
 
     keys = []
     mimes = []
-    r = redis.StrictRedis.from_url(redis_url, charset="utf-8", decode_responses=True)
+    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)
@@ -77,13 +82,13 @@ def read_in_chunks(file_object, chunk_size=1024):
     g = group(analyze_upload.s(k, m, etl) for k, m in zip(keys, mimes))
     return jsonify(g.apply_async().get())
 
-        
+
 @blueprint.route('/api/v1/query')
 @environment('REDIS')
 def index(redis_url):
-    r = redis.StrictRedis.from_url(redis_url, charset="utf-8", decode_responses=True)
+    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)
+    current_app.logger.info(f'Querying for: {iri}')
     if rfc3987.match(iri):
         if not r.exists(iri):
             abort(404)

tsa/tasks.py

@@ -1,41 +1,43 @@
 """Celery tasks invoked from the API endpoints."""
 import json
 import logging
+from urllib.parse import urlparse
+
 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}")
+    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}")
+    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")
+    log.info('System check successful')
 
 
 @celery.task
 def hello():
-    return "Hello world!"
+    return 'Hello world!'
 
 
 @celery.task
 def analyze(iri, etl=True):
     log = logging.getLogger(__name__)
-    log.info(f"Analyzing {iri!s}")
+    log.info(f'Analyzing {iri!s}')
     if etl:
         (transform.s(iri) | poll.s() | inspect.s()).apply_async()
     else:
@@ -45,17 +47,18 @@ def analyze(iri, etl=True):
             r.raise_for_status()
             guess = r.headers.get('content-type')
         g = rdflib.ConjunctiveGraph()
-        log.info(f"Guessing format to be {guess!s}")
+        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
+    exp = 60 * 60  # 1H
     for (s, p, o) in g:
         s = str(s)
         p = str(p)
@@ -73,24 +76,24 @@ def index(g, source_iri, redis_cfg):
         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
+    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) 
+        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()
@@ -99,81 +102,84 @@ def inspect(iri):
 
 @celery.task
 @environment('ETL', 'VIRTUOSO', 'DBA_PASSWORD')
-def transform(iri, etl, virtuoso, dbaPass):
+def transform(iri, etl, virtuoso, dba_pass):
     log = logging.getLogger(__name__)
-    #create pipeline and call to start executions
+    # create pipeline and call to start executions
     # prepare JSON-LD pipeline
 
-    log.info(f"Prepare pipeline for {iri!s}")
+    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}))
+    pipeline = json.dumps(pf.create_pipeline(iri, {'server': p.hostname,
+                                                   'port': 1111,
+                                                   'user': 'dba',
+                                                   'password': dba_pass,
+                                                   'iri': iri}))
 
-    log.info(f"Pipeline:\n{pipeline!s}")
+    log.info(f'Pipeline:\n{pipeline!s}')
 
     # create the pipeline
-    r = requests.post(f"{etl!s}/resources/pipelines", files={'pipeline': 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")
+    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}")
+    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 = requests.post(f'{etl!s}/resources/executions?pipeline={pipeline}')
     r.raise_for_status()
-    log.info(f"Execution trigger result:\n{r.json()!s}")
+    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)
+@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}")
+    log.info(f'Polling {iri!s}')
 
-    r = requests.get(iri + "/overview")
+    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")
+    if j['status']['@id'] == 'http://etl.linkedpipes.com/resources/status/failed':
+        log.error('Execution failed')
 
         try:
-            r = requests.get(iri + "/logs")
+            r = requests.get(iri + '/logs')
             r.raise_for_status()
-            log.error("ETL log:\n" + r.text)
-        except HTTPError as e:
+            log.error('ETL log:\n' + r.text)
+        except requests.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")
+    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 = ""
+        # get result uri
+        result = ''
         return result
 
 
 @celery.task
 @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}")
+    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")
+    log.info(f'Pipeline {iri!s} deleted')
 
 
 class EtlError(Exception):