api.py

import json

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)

    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)

    if 'lat' in request.args or 'lng' in request.args:
        lat = request.args['lat']
        lon = request.args['lon']
        return get_interpolation_data_for_point(lat, lon, columns)

    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

        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)

        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)