Created
October 7, 2020 14:10
-
-
Save isc-rsingh/084e4ecf347696296b214131f825ef4b to your computer and use it in GitHub Desktop.
Python Flask middleware app with InterSystems IRIS as the back end database
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import pyodbc | |
| from flask import Flask | |
| from flask import request | |
| from flask import render_template | |
| import json | |
| app = Flask(__name__) | |
| connection = None | |
| # default route | |
| @app.route('/') | |
| def hello_world(): | |
| return 'Hello, World!' | |
| @app.route('/makeschema', methods=['GET']) | |
| def make_schema(): | |
| try: | |
| cursor = connection.cursor() | |
| sql = 'CREATE TABLE FLASK.SAMPLE(user_id VARCHAR(16), firstname VARCHAR(128), lastname VARCHAR(128) )' | |
| cursor.execute(sql) | |
| connection.commit() | |
| except Exception as exp: | |
| print(exp) | |
| return render_template('/error.html', err=exp) | |
| return 'Successfully created FLASK.SAMPLE table' | |
| @app.route('/loaddata', methods=['POST']) | |
| def load_data(): | |
| d = request.get_data(as_text=True) | |
| personlist = d.split('|') | |
| try: | |
| cursor = connection.cursor() | |
| for person in personlist: | |
| sql = 'INSERT INTO FLASK.SAMPLE (user_id, firstname, lastname) VALUES (' | |
| atts = person.split(',') | |
| sql += "'" + atts[0] + "'" + "," | |
| sql += "'" + atts[1] + "'" + "," | |
| sql += "'" + atts[2] + "'" + ")" | |
| print("Inserting: "+sql) | |
| cursor.execute(sql) | |
| connection.commit() | |
| except Exception as exp: | |
| print(exp) | |
| return render_template('/error.html', err=exp) | |
| return 'Successfully inserted data into FLASK.SAMPLE ' | |
| @app.route('/person/<firstname>') | |
| def get_by_firstname(firstname): | |
| try: | |
| cursor = connection.cursor() | |
| sql = "SELECT * FROM FLASK.SAMPLE WHERE firstname LIKE " + "'" + firstname + "'" | |
| print("SQL: "+sql) | |
| rows = cursor.execute(sql) | |
| resp = [] | |
| for row in rows: | |
| print(row) | |
| r = {"id": row[0], "firstname": row[1], "lastname": row[2]} | |
| resp.append(r) | |
| return json.dumps({ "query": sql, "responses": resp}) | |
| except Exception as exp: | |
| print(exp) | |
| return render_template('/error.html', err=exp) | |
| return 'Successfully inserted data into FLASK.SAMPLE ' | |
| # Get connection details from config file | |
| def get_connection_info(file_name): | |
| # Initial empty dictionary to store connection details | |
| connections = {} | |
| # Open config file to get connection info | |
| with open(file_name) as f: | |
| lines = f.readlines() | |
| for line in lines: | |
| # remove all white space (space, tab, new line) | |
| line = ''.join(line.split()) | |
| # get connection info | |
| connection_param, connection_value = line.split(":") | |
| connections[connection_param] = connection_value | |
| return connections | |
| def run(): | |
| # Retrieve connection information from configuration file | |
| connection_detail = get_connection_info("connection.config") | |
| ip = connection_detail["ip"] | |
| port = int(connection_detail["port"]) | |
| namespace = connection_detail["namespace"] | |
| username = connection_detail["username"] | |
| password = connection_detail["password"] | |
| driver = "{InterSystems ODBC}" | |
| # Create connection to InterSystems IRIS | |
| connection_string = 'DRIVER={};SERVER={};PORT={};DATABASE={};UID={};PWD={}' \ | |
| .format(driver, ip, port, namespace, username, password) | |
| global connection | |
| connection = pyodbc.connect(connection_string) | |
| connection.setdecoding(pyodbc.SQL_CHAR, encoding='utf-8') | |
| connection.setencoding(encoding='utf-8') | |
| print("Connected to InterSystems IRIS") | |
| run() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment