forked from sah0017/DB-Programming-examples
-
Notifications
You must be signed in to change notification settings - Fork 49
/
parameter.py
68 lines (54 loc) · 1.84 KB
/
parameter.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
# an object of WSGI application
from flask import Flask, request
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import text
import os
from dotenv import load_dotenv
load_dotenv()
my_password = str(os.environ.get('PASSWORD'))
app = Flask(__name__) # Flask constructor
db_cred = {
'user': 'root', # DATABASE USER
'pass': my_password, # DATABASE PASSWORD
'host': '127.0.0.1:3306', # DATABASE HOSTNAME
'name': 'sakila' # DATABASE NAME
}
db_uri = f"mysql+pymysql://{db_cred['user']}:{db_cred['pass']}@{db_cred['host']}/{db_cred['name']}"
app.config['SQLALCHEMY_DATABASE_URI'] = db_uri
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# Creating an SQLAlchemy instance
db = SQLAlchemy(app)
# A decorator used to tell the application
# which URL is associated function
@app.route('/hello')
def hello():
return 'HELLO'
# APP ROUTE TO GET RESULTS FOR SELECT QUERY
@app.route('/get_actors', methods=['GET','POST'])
def get_actors():
# GET THE SQLALCHEMY RESULTPROXY OBJECT
with db.engine.begin() as conn:
result = conn.execute(text(request.get_json()['query']))
response = {}
i = 1
# ITERATE OVER EACH RECORD IN RESULT AND ADD IT
# IN A PYTHON DICT OBJECT
for each in result:
response.update({f'Record {i}': list(each)})
i+= 1
return response
@app.route('/actor/<id>',methods=["GET"])
def get_actor(id):
sql = f"SELECT * from actor WHERE actor_id = {id}"
with db.engine.begin() as conn:
result = conn.execute(text(sql))
response = {}
i = 1
# ITERATE OVER EACH RECORD IN RESULT AND ADD IT
# IN A PYTHON DICT OBJECT
for each in result:
response.update({f'Record {i}': list(each)})
i+= 1
return response
if __name__=='__main__':
app.run(port=8000, debug=True)