The Requests Server Code¶

This code in is server.py which you can copy from this directory

cd ~/cs304/
cp -r ~cs304flask/pub/downloads/people_api/ people_api/

Here's the code. It's pretty short; fewer than 150 lines long. Most of the code should be familiar to you. However, I decided that allowing POST to the WMDB could cause all kinds of bogus data to be inserted, so the inserted data goes to an in-memory dictionary, and the GET requests combine a search of the SQL database with the inserted data.

'''Flask/PyMySQL app to list the people in the WMDB providing a JSON
API, suitable for using with the requests module.

Scott D. Anderson
revised Fall 2019 for Python3
revised Spring 2020 for new dbi module
Fall 2023 adapted for the JSON API
'''

from flask import (Flask, url_for, render_template, request, session, jsonify)
import secrets
from datetime import datetime
import cs304dbi as dbi
import servertime
import people

app = Flask(__name__)
app.secret_key = secrets.token_hex()

@app.route('/')
def hello_world():
    now = servertime.now()
    return f'provides a API to the WMDB. Time is {now}'

## Global values

force_auth = False

# default username and password. For better security, don't use a default

auth_username = 'me'
auth_password = 'secret'

## I don't want people updating the WMDB, so I'll put posted entries in this.
new_people_db = {}
    
## ================================================================

def new_folk():
    sorted_keys = sorted(new_people_db.keys())
    return [ new_people_db[key]
             for key in sorted_keys ]

def new_folk_in_month(month):
    sorted_keys = sorted(new_people_db.keys())
    folk = []
    for nm in sorted_keys:
        person = new_people_db[nm]
        bdate = person['birthdate']
        if bdate is not None and bdate.month == month:
            folk.append(person)
    return folk

def to_date(date_str):
    '''Parse an ISO 8601 date'''
    try:
        return datetime.strptime(date_str, '%Y-%m-%d')
    except:
        return None

def insert_person(info):
    '''Inserts a new person into the in-memory database from a dictionary
of values. The argument could be a Flask immutable_multidict.'''
    # there's no error checking here
    nm = info['nm']
    person = {'nm': nm,
              'name': info['name'],
              # the following are optional, hence .get(), defaulting to None
              'birthdate': to_date(info.get('birthdate')),
              'addedby': info.get('addedby')}
    new_people_db[nm] = person
    
## ================================================================

@app.route('/people/')
def people_list():
    conn = dbi.connect()
    folk = people.get_people(conn)
    folk.extend(new_folk())
    return jsonify({'error': False, 'data': folk})
    
@app.route('/people-born-in-month/<int:month>')
def people_born_in_month(month):
    conn = dbi.connect()
    folk = people.people_born_in_month(conn,month)
    folk.extend(new_folk_in_month(month))
    return jsonify({'error': False, 'data': folk})

@app.route('/people-noauth/', methods = ['POST'])
def add_person_noauth():
    '''add a person to the database via a posted form'''
    if force_auth:
        return jsonify({'error': 'Must use auth endpoint'})
    insert_person(request.form)
    return jsonify({'error': False, 'data': 'person added'})

@app.route('/login/', methods = ['POST'])
def login():
    print('client sent credentials', request.form.get('username'), request.form.get('password'))
    if (request.form.get('username') == auth_username and
        request.form.get('password') == auth_password):
        session['auth'] = True
        print('login success')
        return jsonify({'error': False, 'data': 'successfully logged in'})
    else:
        print('login failure')
        return jsonify({'error': 'incorrect username/password'})

@app.route('/people-auth/', methods = ['POST'])
def add_person_auth():
    '''add a person to the database via a posted form'''
    if not session.get('auth'):
        return jsonify({'error': 'not logged in'})
    insert_person(request.form)
    return jsonify({'error': False, 'data': 'person added'})


if __name__ == '__main__':
    import os
    uid = os.getuid()
    dbi.conf('wmdb')

    print('If you force authentication, the example client app will fail.')
    ans = input('Force authentication? (y/n) (default: y) ') or 'y'
    force_auth = (ans == 'y')
    auth_username = input(f'username: (default: {auth_username}) ') or auth_username
    auth_password = input(f'password: (default: {auth_password}) ') or auth_password
    app.debug = True
    app.run('0.0.0.0',uid)