-
Notifications
You must be signed in to change notification settings - Fork 1
/
end_points.py
114 lines (81 loc) · 3.77 KB
/
end_points.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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
import logging
from flask import request, jsonify
from database import models
import util
from sqlalchemy import inspect
def row2dict(row):
"""
Get dict for row.
Note mapped rows are *logical*, not physical db table/column names.
* see Category, Order.ShipZip
https://docs.sqlalchemy.org/en/20/orm/mapping_styles.html#inspection-of-mapped-instances
Args:
row (_type_): row instance (mapped object)
Returns:
_type_: dict of attr names/values
"""
mapper = inspect(row.__class__).mapper
return {
c.key: str(getattr(row, c.key))
for c in mapper.attrs # not: for c in row.__table__.columns
}
def flask_events(app, db):
"""
Illustrate flask events
"""
app_logger = logging.getLogger(__name__)
@app.route('/hello_world')
def hello_world():
"""
Illustrates simplest possible endpoint, with url args.
The url suffix is specified in the annotation.
Test it with: http://localhost:8080/hello_world?user=Basic_App
Returns:
json : a simple string
"""
user = request.args.get('user') # obtain URL argument from Flask via built-in request object
app_logger.info(f'{user}')
return jsonify({"result": f'hello, {user}'}) # the api response (in json)
@app.route('/order', methods=['GET']) # tell Flask: call this function when /order request occurs
def order():
"""
End point to return a nested result set response, from related database rows
Illustrates:
1. Obtain URL argument from Flask
2. Read data from SQLAlchemy, and related data (via foreign keys)
3. Restructure row results to desired json (e.g., for tool such as Sencha)
4. Use Flask to return nested response json
Test:
http://localhost:8080/order?Id=10643
curl -X GET "http://localhost:8080/order?Id=10643"
"""
# 1. Obtain URL argument from Flask
order_id = request.args.get('Id')
# 2. Read data from SQLAlchemy
order = db.session.query(models.Order).\
filter(models.Order.Id == order_id).one()
app_logger.info(f'\n Breakpoint - examine order in debugger \n')
# 3. Restructure row results - format as result_std_dict
result_std_dict = util.format_nested_object(row2dict(order)
, remove_links_relationships=False)
result_std_dict['Customer_Name'] = order.Customer.CompanyName # eager fetch
result_std_dict['OrderDetailListAsDicts'] = []
for each_order_detail in order.OrderDetailList: # SQLAlchemy related data access
each_order_detail_dict = util.format_nested_object(row=row2dict(each_order_detail)
, remove_links_relationships=False)
each_order_detail_dict['ProductName'] = each_order_detail.Product.ProductName
result_std_dict['OrderDetailListAsDicts'].append(each_order_detail_dict)
# 4. Use Flask to return nested response json (Flask jsonifies dict)
return result_std_dict # rest response
@app.route('/stop')
def stop(): # test it with: http://localhost:8080/stop?msg=API stop - Stop Basic App Server
"""
Use this to stop the server from the Browser.
See: https://stackoverflow.com/questions/15562446/how-to-stop-flask-application-without-using-ctrl-c
"""
import os, signal
msg = request.args.get('msg')
app_logger.info(f'\nStopped server: {msg}\n')
os.kill(os.getpid(), signal.SIGINT)
return jsonify({ "success": True, "message": "Server is shutting down..." })
logging.info("\n\n..Basic App, exposing end points: hello_world, order, stop\n")