dm-run-server

#!/usr/bin/env python2.7
# -*- coding: utf-8 -*-

import getopt
import json
import mimetypes
import os
import re
import sys

from flask import Flask, request, Response, jsonify

from hecatoncheir import DbProfilerFormatter
from hecatoncheir import DbProfilerVerify
from hecatoncheir import db
from hecatoncheir import logger as log
from hecatoncheir.businessglossary import GlossaryTerm, get_bg_term
from hecatoncheir.datamapping import get_datamap_items
from hecatoncheir.msgutil import gettext as _
from hecatoncheir.repository import Repository
from hecatoncheir.schema import Schema2
from hecatoncheir.table import Table2
from hecatoncheir.tag import Tag2
from hecatoncheir.validation import ValidationRule, get_validation_rules

app = Flask(__name__)


def open_repo():
    db.creds = db.parse_connection_string(os.environ["DBPROF_REPOFILE"])
    db.connect()
    return True


def get_glossary_terms():
    terms = []
    for t in GlossaryTerm.find():
        terms.append(get_bg_term(t.term))
    return terms


@app.route("/")
@app.route("/index.html")
def index():
    open_repo()
    terms = get_glossary_terms()

    tables_all = []
    for tab in Table2.find():
        tables_all.append(tab.data)
    schemas = []
    for s in Schema2.findall():
        # [dbname,schemaname,num_of_tables,desc]
        schemas.append([s.database_name, s.schema_name,
                        s.num_of_tables, s.description])
    tags = []
    for tag in Tag2.findall():
        if tag.num_of_tables == 0:
            tag.destroy()
            continue
        tags.append([tag.label, tag.num_of_tables, tag.description])

    reponame = re.sub('.*/', '', db.creds['dbname'])
    reponame = re.sub('\..*$', '', reponame)
    html = DbProfilerFormatter.to_index_html(tables_all, schemas=schemas,
                                             tags=tags,
                                             show_validation='both',
                                             reponame=reponame,
                                             glossary_terms=terms,
                                             max_panels=6,
                                             editable=True).encode('utf-8')

    return html


@app.route("/index-tags.html")
def index_tags():
    open_repo()
    terms = get_glossary_terms()

    tables_all = []
    for tab in Table2.find():
        tables_all.append(tab.data)
    tags = []
    for tag in Tag2.findall():
        tags.append([tag.label, tag.num_of_tables, tag.description])

    reponame = re.sub('.*/', '', db.creds['dbname'])
    reponame = re.sub('\..*$', '', reponame)
    html = DbProfilerFormatter.to_index_html(tables_all,
                                             tags=tags,
                                             reponame=reponame,
                                             glossary_terms=terms,
                                             max_panels=99,
                                             editable=True).encode('utf-8')

    return html


@app.route("/index-schemas.html")
def index_schemas():
    open_repo()
    terms = get_glossary_terms()

    tables_all = []
    for tab in Table2.find():
        tables_all.append(tab.data)

    # [[dbname, schema, num_of_tables, desc],
    #  [dbname, schema, num_of_tables, desc], ...]
    schemas = []
    for s in Schema2.findall():
        schemas.append([s.database_name, s.schema_name,
                        s.num_of_tables, s.description])

    reponame = re.sub('.*/', '', db.creds['dbname'])
    reponame = re.sub('\..*$', '', reponame)
    html = DbProfilerFormatter.to_index_html(tables_all,
                                             schemas=schemas,
                                             reponame=reponame,
                                             glossary_terms=terms,
                                             max_panels=99,
                                             editable=True).encode('utf-8')

    return html


@app.route("/<db>.<schema>.html")
def index_schema(db, schema):
    open_repo()
    terms = get_glossary_terms()

    s = Schema2.find(db, schema)
    tables = []
    for tab in Table2.find(database_name=db, schema_name=schema):
        tables.append(tab.data)

    html = DbProfilerFormatter.to_index_html(
        tables,
        comment=s.comment,
        schemas=[[s.database_name, s.schema_name,
                  s.num_of_tables, s.description]],
        reponame=schema, glossary_terms=terms,
        editable=True).encode('utf-8')

    return html


@app.route("/tag-<tag>.html")
def index_tag(tag):
    open_repo()
    terms = get_glossary_terms()

    tables = []
    for tab in Table2.find(tag=tag):
        tables.append(tab.data)

    t = Tag2.find(tag)
    assert t
    html = DbProfilerFormatter.to_index_html(tables,
                                             comment=t.comment,
                                             tags=[[tag, t.num_of_tables,
                                                    t.description]],
                                             reponame=tag,
                                             glossary_terms=terms,
                                             editable=True).encode('utf-8')

    return html


@app.route("/validation-<status>.html")
def index_validation(status):
    open_repo()
    terms = get_glossary_terms()

    tables = []
    for t in Table2.find():
        tab = t.data
        (valid, invalid) = DbProfilerVerify.verify_table(tab)
        if status == 'invalid' and invalid > 0:
            tables.append(tab)
        elif status == 'valid' and invalid == 0 and valid > 0:
            tables.append(tab)
    html = DbProfilerFormatter.to_index_html(tables,
                                             show_validation=status,
                                             reponame=status,
                                             glossary_terms=terms,
                                             editable=True).encode('utf-8')

    return html


@app.route("/<db>.<schema>.<table>.html")
def table(db, schema, table):
    open_repo()
    terms = get_glossary_terms()

    table_data = Table2.find(db, schema, table)[0].data
    datamap = get_datamap_items(db, schema, table)
    validation_rules = get_validation_rules(db, schema, table)
    html = DbProfilerFormatter.to_table_html(table_data,
                                             validation_rules=validation_rules,
                                             datamapping=datamap,
                                             glossary_terms=terms,
                                             editable=True).encode('utf-8')

    return html


@app.route("/glossary.html")
def glossary():
    open_repo()
    terms = get_glossary_terms()

    html = DbProfilerFormatter.to_glossary_html(glossary_terms=terms,
                                                editable=True).encode('utf-8')
    return html


@app.route("/static/<filename>")
@app.route("/static/<dir1>/<filename>")
@app.route("/static/<dir2>/<dir1>/<filename>")
def staticfile(filename, dir1=None, dir2=None):
    static_dir = DbProfilerFormatter.get_default_template_path()
    if dir1:
        filename = dir1 + '/' + filename
    if dir2:
        filename = dir2 + '/' + filename
    out = ""
    for l in open(static_dir + '/static/' + filename):
        out = out + l

    mime = mimetypes.guess_type(filename)
    return Response(out, mimetype=(mime[0] if mime[0]
                                   else 'application/octet-stream'))


# --------------------------------------------------
# REST API
# --------------------------------------------------
@app.route("/api/metadata/<db>.<schema>.<table>")
def api_metadata(db, schema, table):
    open_repo()
    data = Table2.find(db, schema, table)[0].data

    return json.dumps(data, indent=2)


@app.route("/api/table/<db>.<schema>.<table>/table_info",
           methods=['GET', 'POST'])
def api_table_owner(db, schema, table):
    open_repo()
    tab = Table2.find(db, schema, table)[0]
    data = tab.data
    if request.method == 'POST':
        data['owner'] = request.form['owner']
        data['comment'] = request.form['comment']
        data['tags'] = []
        for t in request.form['tags'].replace(' ', '').split(','):
            if len(t) > 0:
                data['tags'].append(t)
        tab.update()
        for tag in data['tags']:
            if not Tag2.find(tag):
                Tag2.create(tag)
        return json.dumps({'owner': data.get('owner', ''),
                           'comment': data.get('comment', ''),
                           'tags': data.get('tags', [])})
    else:
        return json.dumps({'owner': data.get('owner', ''),
                           'comment': data.get('comment', ''),
                           'tags': data.get('tags', [])})


@app.route("/api/comment/<db>.<schema>.<table>.<column>",
           methods=['GET', 'POST'])
def api_column_comment(db, schema, table, column):
    open_repo()
    tab = Table2.find(db, schema, table)[0]
    data = tab.data
    col = None
    for c in data['columns']:
        if c['column_name'] == column:
            col = c
    if request.method == 'POST':
        col['comment'] = request.form['comment']
        tab.update()
        return json.dumps({'comment': col['comment']})
    else:
        return json.dumps({'comment': col['comment']})


def api_error(code, msg):
    r = jsonify({'message': msg, 'status': 'error'})
    r.status_code = code
    return r


def api_response(code, content):
    assert isinstance(content, dict)
    r = jsonify(content)
    r.status_code = code
    return r


@app.route("/api/validation", methods=['GET'])
def api_validation_get_all():
    open_repo()
    a = ['id', 'database_name', 'schema_name', 'table_name', 'column_name',
         'description', 'rule', 'param', 'param2']
    data = []
    try:
        rr = get_validation_rules(request.args.get('database_name'),
                                  request.args.get('schema_name'),
                                  request.args.get('table_name'))
        for r in rr:
            assert len(a) == len(r)
            d = {}
            for k, v in zip(a, r):
                d[k] = v
            data.append(d)
    except Exception as e:
        return api_error(400, 'exception caught.')

    resp = {'status': 'success',
            'data': data}
    return api_response(201, resp)


@app.route("/api/validation", methods=['POST'])
def api_validation_create():
    if not request.data:
        return api_error(400, 'the request data is empty.')

    req = None
    try:
        req = json.loads(request.data)
    except ValueError as e:
        return api_error(400, 'incorrect data format.')

    open_repo()
    id = None
    try:
        r = ValidationRule.create(
            req['database_name'], req['schema_name'], req['table_name'],
            req['column_name'], req['description'], req['rule'],
            req.get('param'), req.get('param2'))
        id = r.id
    except Exception as ex:
        id = None

    if not id:
        return api_error(400, 'could not register a validation rule.')
    resp = {'id': id,
            'status': 'success'}
    return api_response(201, resp)


@app.route("/api/validation/<id>", methods=['GET'])
def api_validation_get(id):
    open_repo()
    r = None
    try:
        id = int(id)
        r = ValidationRule.find(id_=id)
        if r is None:
            return api_error(400, 'rule id %d not found.' % id)
    except Exception as e:
        return api_error(400, 'exception caught.')

    assert r.id == id
    resp = {'id': r.id,
            'status': 'success',
            'database_name': r.database_name,
            'schema_name': r.schema_name,
            'table_name': r.table_name,
            'column_name': r.column_name,
            'description': r.description,
            'rule': r.rule,
            'param': r.param,
            'param2': r.param2}
    return api_response(201, resp)


@app.route("/api/validation/<id>", methods=['PUT'])
def api_validation_put(id):
    if not request.data:
        return api_error(400, 'the request data is empty.')
    req = None
    try:
        req = json.loads(request.data)
    except ValueError as e:
        return api_error(400, 'incorrect data format.')

    open_repo()
    try:
        id = int(id)
        v = ValidationRule.find(id_=id)
        assert len(v) == 1
        v[0].database_name = req['database_name']
        v[0].schema_name = req['schema_name']
        v[0].table_name = req['table_name']
        v[0].column_name = req['column_name']
        v[0].description = req['description']
        v[0].rule = req['rule']
        v[0].param = req.get('param')
        v[0].param2 = req.get('param2')
        if not v[0].update():
            return api_error(400, 'rule %d could not updated.' % id)
    except Exception as e:
        return api_error(400, 'exception caught.')

    resp = {'id': id,
            'status': 'success'}
    return api_response(201, resp)


@app.route("/api/validation/<id>", methods=['DELETE'])
def api_validation_delete(id):
    open_repo()
    try:
        id = int(id)
        v = ValidationRule.find(id_=id)
        assert len(v) == 1
        if not v[0].destroy():
            return api_error(400, 'no rule deleted.')
    except Exception as e:
        return api_error(400, 'exception caught.')

    resp = {'id': id,
            'status': 'success'}
    return api_response(201, resp)


def usage():
    print '''
Usage: %s [repo file | connection string] [port]

Options:
    --help                     Print this help.

''' % os.path.basename(sys.argv[0])


if __name__ == "__main__":
    try:
        opts, args = getopt.getopt(sys.argv[1:], "",
                                   ["help", "debug"])
    except getopt.GetoptError as err:
        log.error(unicode(err))
        usage()
        sys.exit(1)

    debug = False

    for o, a in opts:
        if o in ("--debug"):
            debug = True
        elif o in ("--help"):
            usage()
            sys.exit(0)
        else:
            log.error("unexpected option. internal error.")
            sys.exit(1)

    if len(args) < 1:
        usage()
        sys.exit(1)

    os.environ["DBPROF_REPOFILE"] = args[0]

    port = 8080
    if len(args) == 2:
        try:
            port = int(args[1])
        except Exception as e:
            log.error(_("%s is not a correct port number.") % args[1])
            sys.exit(1)

    app.run(host='0.0.0.0', port=port)