api.py

import psycopg2
from flask import Flask, jsonify, request
from psycopg2 import sql
import configparser
from GetAverageData import get_interpolation_data_for_point

cfg = configparser.ConfigParser()
cfg.read('../config.ini')
assert "POSTGRES" in cfg, "missing POSTGRES in config.ini"
param_postgres = cfg["POSTGRES"]

app = Flask(__name__)
app.config['ENV'] = "development"
app.config['DEBUG'] = True
# Testing: Exceptions are propagated rather than handled by the the app’s error handlers. Extensions may also change their behavior to facilitate easier testing. You should enable this in your own tests.
app.config['TESTING'] = False


@app.route('/', methods=['GET'])
def index():
    columns = sql.SQL(' * ')    # columns to be queried (e.g. years)
    wheres = sql.SQL('')        # where filters
    values = ''                 # Used in second parameter of cursor.execute() (Avoids SQL injection)

    # Filter for getting specific years
    if 'years' in request.args:
        years = request.args['years'].split(',')
        years_clean = []

        for year in years:
            years_clean.append('"' + year + '"')
        years_clean = str(years_clean).replace('[', '').replace(']', '').replace("'", "").replace('\n', '').replace(' ', '')

        columns = sql.SQL(years_clean)

    # Filter for getting only real measurent data without any interpolated points
    if 'no-interpol' in request.args:
        wheres = wheres + (sql.SQL("AND file IS NOT NULL "))

    # Filter for specific coordinates
    if 'lat' in request.args or 'lng' in request.args:
        lat = request.args['lat']
        lon = request.args['lon']
        with psycopg2.connect(database=param_postgres["dbName"], user=param_postgres["user"], password=param_postgres["password"], host=param_postgres["host"], port=param_postgres["port"]) as connection:
            with connection.cursor() as cursor:
                return get_interpolation_data_for_point(lat, lon, columns, cursor)

    else:
        if 'id' in request.args:
            station_id = request.args['id']
            wheres = wheres + (sql.SQL("AND id = {values} ").format(column=sql.Identifier('stations', 'id'), values=sql.Placeholder()))
            for n in [int(station_id)]:
                values = (*values, n)  # adding n to existing tuple

        # Just for development
        # if 'country' in request.args:
        #     country = request.args['country']
        #     wheres = wheres + (sql.SQL("AND LOWER({column}) LIKE {values} ").format(column=sql.Identifier('stations', 'country'), values=sql.Placeholder()))
        #     for n in [country]:
        #         values = (*values, n)  # adding n to existing tuple

        # Composing query
        query = sql.SQL("SELECT array_to_json(array_agg(row_to_json(t))) from ("
                        "SELECT id, {} FROM stations "
                        "WHERE lon IS NOT NULL "    # Unnecessary filter, just so the real filters can always be written with AND
                        "{} "
                        ") t;").format(columns, wheres)

        # Actual query execution
        with psycopg2.connect(database=param_postgres["dbName"], user=param_postgres["user"], password=param_postgres["password"], host=param_postgres["host"], port=param_postgres["port"]) as connection:
            with connection.cursor() as cursor:
                # print(query.as_string(cursor))
                # print(values)
                cursor.execute(query, values)
                results = cursor.fetchall()[0][0]
                return jsonify(results)


app.run(host='127.0.0.1', port=42000)