Skip to content

Instantly share code, notes, and snippets.

@pwin

pwin/REST-API.py

Last active October 18, 2024 13:43
Show Gist options
  • Save pwin/8391942 to your computer and use it in GitHub Desktop.
Save pwin/8391942 to your computer and use it in GitHub Desktop.
Download ZIP
A REST-ful API script using Bottle.py
Raw
make_table.tpl
%#template to generate a HTML table from a list of tuples (or list of lists, or tuple of tuples or ...)
%#includes a converter for strings to utf-8
%_str = lambda x: x if not isinstance(x, str) else x.decode('utf-8','replace')
<html>
<head>
<title>Output from API Call</title>
</head>
<body>
<table border="1">
<thead><tr>
%for c in cols:
<td>{{c[0]}}</td>
%end
</tr>
</thead>
<tbody>
%for row in rows:
<tr>
%for col in row:
<td>{{_str(col)}}</td>
%end
</tr>
%end
</tbody>
</table>
</body>
Raw
REST-API.py
#!/usr/bin/env python
"""
REST-API.py
The MIT License (MIT)
Copyright (c) 2014 P.Winstanley, C.Taylor
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
"""
from bottle import Bottle, route, run, template, request, response, error, abort, static_file
import pyodbc
import cStringIO as StringIO
import csv
import json
import dict2xml
import ast
import sys
reload(sys)
sys.setdefaultencoding('utf8')
import logging
debug = False
# MIME Types
MIME_JSON="application/json"
MIME_HTML="text/html"
MIME_TEXT="text/plain"
MIME_XML="application/xml"
ODBC = "{Connection String}"
app = Bottle()
#Enable CORS
@app.hook('after_request')
def enable_cors():
"""
You need to add some headers to each request.
Don't use the wildcard '*' for Access-Control-Allow-Origin in production.
"""
response.headers['Access-Control-Allow-Origin'] = '*'
response.headers['Access-Control-Allow-Methods'] = 'GET, OPTIONS'
response.headers['Access-Control-Allow-Headers'] = 'Origin, Accept, Content-Type, X-Requested-With, X-CSRF-Token'
def row2utf8(row):
r = [x if not isinstance(x, str) else x.decode('utf-8','replace') for x in row]
return tuple(r)
##Info Page
@app.route('/')
@app.route('/info')
def info():
return template('info')
## connection to a database and also return first result from SQL query
@app.route('/database', method="GET")
@app.route('/database/<name>', method="GET")
def database1(name=None):
cnxn = pyodbc.connect(ODBC)
cursor = cnxn.cursor()
if name:
cursor.execute("select * from dbo.%s" % (name))
else:
cursor.execute("select * from dbo.services")
row = cursor.fetchone()
if row:
return str(row)
## connection to a database return names of resources (areas, locations, services etc) SQL query
@app.route('/v1/database/name', method='GET')
@app.route('/v1/database/name/<table>', method='GET')
@app.route('/v1/database/name/<table>.<ext>', method='GET')
@app.route('/v1/database/name/<table>/<id>', method='GET')
@app.route('/v1/database/name/<table>/<id>.<ext>', method='GET')
def database(table=None, id=None, ext='html', fields='*'):
if request.query_string:
fields = ','.join(request.query.fields.split(',')) or '*'
translate_id_name = {'areas':'area_id',
'location_type':'loctype_id',
'service_type':'servtype_id',
'divisions':'div_id',
'services':'service_id',
'locations':'loc_id',
'loc_services':'loc_serv_id'
}
accept = request.environ.get('HTTP_ACCEPT')
if table:
#convert the table name in the API route to the database table name for the query
id_name = translate_id_name[table]
else:
name = None
abort(404, 'No table name given')
if id:
record = ' where %s = %s' % (id_name, id)
else:
record = ''
cnxn = pyodbc.connect(ODBC)
cursor = cnxn.cursor()
if table:
sql = "select %s from dbo.%s %s;" % (fields, table, record)
else:
abort(404, "No records found")
#get the accept mime type from the header
accept1 = accept.split(',')[0]
rows = execute_sql(cursor, sql)
if len(rows) == 0:
abort(404, "No records found")
if (MIME_JSON in accept1) or ('json' in ext.lower()):
if debug: print('it is json')
response.content_type=MIME_JSON
columns = [column[0] for column in cursor.description]
results = []
for row in rows:
row1 = row2utf8(row)
results.append(dict(zip(columns, row1)))
output = {"Records":results}
elif (MIME_TEXT in accept1) or ('csv' in ext):
if debug: print('it is txt')
response.content_type=MIME_TEXT
csvfile = StringIO.StringIO()
csvwriter = csv.writer(csvfile)
csvwriter.writerow([column[0] for column in cursor.description])
for row in rows:
try:
row1 = row2utf8(row)
csvwriter.writerow(row1)
except Exception, e:
abort(500, "Error: " + str(e))
output = read_and_flush(csvfile)
csvfile.close()
elif (MIME_XML in accept1) or ('xml' in ext.lower()):
response.content_type=MIME_XML
columns = [column[0] for column in cursor.description]
results = []
for row in rows:
row1 = row2utf8(row)
results.append(dict(zip(columns, row1)))
json_obj = {"Records":{"Record":results}}
output = dict2xml.dict2xml(json_obj)
else:
response.content_type=MIME_HTML
output = template('make_table', rows=rows, cols=cursor.description)
cursor.close()
return output
##########Some Helper Functions etc #########
def read_and_flush(csvfile):
csvfile.seek(0)
data = csvfile.read()
csvfile.seek(0)
csvfile.truncate()
if debug: print(data)
return data
def execute_sql(cursor, sql):
return cursor.execute(sql).fetchall()
# log the headers
def dumpHeaders():
print('---- HEADERS ----')
for k in request.headers.keys():
print("%15s : %s" %(k, request.headers.get(k)))
print('\n')
@app.error(404)
def error404(code):
return "404 error. Results Not Found"
@app.error(500)
def error404(code):
return "500 error. Server Error"
if __name__ == "__main__":
run(app, reloader=True, host='localhost', port=8080, debug=True)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment