Skip to content

Instantly share code, notes, and snippets.

@sferich888
Last active July 26, 2018 00:51
Show Gist options
  • Save sferich888/a790ea8cd8c8bf2c7a2c52882f88f695 to your computer and use it in GitHub Desktop.
Save sferich888/a790ea8cd8c8bf2c7a2c52882f88f695 to your computer and use it in GitHub Desktop.
Mock API Server (flask) and Simply 'requests' Client (with threading)
import requests, argparse, timeit
from concurrent.futures import ThreadPoolExecutor
### Globals
""" These are simply defined, and the values are not used.
The values are set by argparse at the start of the script.
"""
server = ''
port = 0
def get_pronoun(name):
r = requests.get('http://{}:{}/kid/{}'.format(server, port, name))
if r.json()['sex'] == 'female':
return 'her'
elif r.json()['sex'] == 'male':
return 'his'
else:
raise Error("Can't determine pronoun.")
def get_age(name):
r = requests.get('http://{}:{}/age/{}'.format(server, port, name))
return r.json()['age']
def get_full_name(name):
r = requests.get('http://{}:{}/full_name/{}'.format(server, port, name))
return r.json()['full_name']
def till_bday(name):
r = requests.get('http://{}:{}/birthday/{}'.format(server, port, name))
return r.json()['days_till']
def main():
tp = ThreadPoolExecutor(args.threads)
def print_message(child):
print("{} is {}, and has {} days till {} birthday.".format(
get_full_name(child['first_name']), get_age(child['first_name']),
till_bday(child['first_name']), get_pronoun(child['first_name'])))
r = requests.get('http://{}:{}/children'.format(server, port))
tp.map(print_message, r.json()['children'], chunksize=1)
tp.shutdown(wait=True)
if __name__ == '__main__':
parser = argparse.ArgumentParser(description="Simple request client.")
parser.add_argument("-p", "--port", help="Change the port for the server.",
action="store", default=5000)
parser.add_argument("-s", "--server", help="Change the server identifier.",
action="store", default='127.0.0.1')
parser.add_argument("-t", "--time", help="Time the client.",
action="store_true", default=False)
parser.add_argument("-c", "--threads", help="Time the client.",
action="store", type=int, default=4)
args = parser.parse_args()
if args.server:
server = args.server
if args.port:
port = args.port
if args.time:
t = timeit.Timer(lambda: main())
print("- Time Taken: {}".format(t.timeit(number=1)))
else:
main()
from flask import Flask, request, Response
import argparse, json, datetime
app = Flask(__name__)
### Mock Data
child1 = {'first_name': 'First1', 'middle_name': 'Middle1', 'last_name': 'Last1',
'sex': 'female', 'birthday': datetime.date(2006, 10, 4).isoformat()}
child2 = {'first_name': 'First2', 'middle_name': 'Middle2', 'last_name': 'Last2',
'sex': 'female', 'birthday': datetime.date(1996, 6, 20).isoformat()}
child3 = {'first_name': 'First3', 'middle_name': 'Middle3', 'last_name': 'Last3',
'sex': 'male', 'birthday': datetime.date(2002, 7, 15).isoformat()}
children = [child1, child2, child3]
### Custom Error
class ChildNotFoundError(Exception):
def __init__(self, message):
super().__init__(message)
### General Functions
def known_kid(first_name):
for kid in children:
if first_name.lower() == kid['first_name'].lower():
return kid
raise ChildNotFoundError("Missing Child!")
def children_of(sex):
kids = []
for kid in children:
if sex == kid['sex']:
kids.append(kid)
return kids
def calculate_age(birthday):
today = datetime.date.today()
dob = datetime.date(*[int(i) for i in birthday.split("-")])
return today.year - dob.year - ((today.month, today.day) < (dob.month, dob.day))
def days_till(birthday):
today = datetime.date.today()
diff = birthday - today
if diff.days >= 0:
return diff.days
else:
return days_till(datetime.date(today.year + 1, birthday.month, birthday.day))
### Mock API Routes
@app.route('/children', methods=['GET'])
def get_children():
ret = {"children": children}
resp = Response(response = json.dumps(ret),
status = 200,
mimetype = "application/json")
return resp
@app.route('/daughters', methods=['GET'])
@app.route('/sons', methods=['GET'])
def get_kids_of():
data = {}
if 'daughters' in str(request.url_rule):
kids = children_of('female')
data = {'daughters': kids}
elif 'sons' in str(request.url_rule):
kids = children_of('male')
data = {'sons': kids}
else:
# Somehow we had a request for a trangender child.
# Well call them a teapot.
return Response(status = 418)
resp = Response(response = json.dumps(data),
status = 200,
mimetype = "application/json")
return resp
@app.route('/kid/<first_name>', methods=['GET'])
def get_kid(first_name):
resp = None
try:
kid = known_kid(first_name)
resp = Response(response = json.dumps(kid),
status = 200, mimetype = "application/json")
except ChildNotFoundError as cnf:
resp = Response(response = json.dumps({'error': {'code': 'E404',
'message': "I don't know this kid! How am I going to tell details about it?"}}),
status = 404, mimetype = "application/json")
return resp
@app.route('/full_name/<first_name>', methods=['GET'])
def get_name(first_name):
resp = None
try:
kid = known_kid(first_name)
full_name = '{} {} {}'.format(kid['first_name'], kid['middle_name'],
kid['last_name'])
resp = Response(response = json.dumps({'full_name': full_name}),
status = 200, mimetype = "application/json")
except ChildNotFoundError as cnf:
resp = Response(response = json.dumps({'error': {'code': 'E404',
'message': "I don't know this kid! How am I going to tell you its full name?"}}),
status = 404, mimetype = "application/json")
return resp
@app.route('/age/<first_name>', methods=['GET'])
def get_age(first_name):
resp = None
if args.slow:
import time
time.sleep(1)
try:
kid = known_kid(first_name)
resp = Response(response = json.dumps({'age': calculate_age(kid['birthday'])}),
status = 200, mimetype = "application/json")
except ChildNotFoundError as cnf:
resp = Response(response = json.dumps({'error': {'code': 'E404',
'message': "I don't know this kid! How am I going to tell you its age?"}}),
status = 404, mimetype = "application/json")
return resp
@app.route('/birthday/<first_name>', methods=['GET'])
def get_birthday(first_name):
resp = None
if args.slow:
import time
time.sleep(1)
try:
kid = known_kid(first_name)
resp = Response(response = json.dumps({'birthday': kid['birthday'],
'days_till': days_till(datetime.date(datetime.date.today().year,
*[int(i) for i in kid['birthday'].split("-")[1:3]])),
'new_age': calculate_age(kid['birthday'])+1}),
status = 200, mimetype = "application/json")
except ChildNotFoundError as cnf:
resp = Response(response = json.dumps({'error': {'code': 'E404',
'message': "I don't know this kid! How am I going to tell you its age?"}}),
status = 404, mimetype = "application/json")
return resp
### Server Functions
def get_routes():
output = []
for rule in app.url_map.iter_rules():
options = {}
for arg in rule.arguments:
options[arg] = "[{0}]".format(arg)
methods = ','.join(rule.methods)
line = "{:30s} {:20s}".format(str(rule), methods)
output.append(line)
return sorted(output)
if __name__ == '__main__':
parser = argparse.ArgumentParser(description="Simple Flask API for testing")
parser.add_argument("-v", "--verbose", help="Turn on debug mode.",
action="store_true", default=False)
parser.add_argument("-p", "--port", help="Change the port for the server.",
action="store", default=5000)
parser.add_argument("-i", "--host", help="Change the ip/host for the server.",
action="store", default='127.0.0.1')
parser.add_argument("-l", "--list", help="List routes/api endpoints.",
action="store_true", default=False)
parser.add_argument("-s", "--slow", help="List routes/api endpoints.",
action="store_true", default=False)
args = parser.parse_args()
if args.list:
import pprint
print(" * {:30s}".format("Data"))
pp = pprint.PrettyPrinter(indent=2)
pp.pprint(children)
print(" * {:30s} {:20s}".format("Routes", "Methods"))
for line in get_routes():
print(" - {}".format(line))
if args.slow:
print(" *************************")
print(" ***SLOW MODE ACTIVATED***")
print(" *************************")
app.run(host=args.host, port=args.port, debug=args.verbose)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment