-
-
Save jamescalam/0b309d275999f9df26fa063602753f73 to your computer and use it in GitHub Desktop.
| from flask import Flask | |
| from flask_restful import Resource, Api, reqparse | |
| import pandas as pd | |
| import ast | |
| app = Flask(__name__) | |
| api = Api(app) | |
| class Users(Resource): | |
| def get(self): | |
| data = pd.read_csv('users.csv') # read local CSV | |
| data = data.to_dict() # convert dataframe to dict | |
| return {'data': data}, 200 # return data and 200 OK | |
| def post(self): | |
| parser = reqparse.RequestParser() # initialize | |
| parser.add_argument('userId', required=True) # add args | |
| parser.add_argument('name', required=True) | |
| parser.add_argument('city', required=True) | |
| args = parser.parse_args() # parse arguments to dictionary | |
| # read our CSV | |
| data = pd.read_csv('users.csv') | |
| if args['userId'] in list(data['userId']): | |
| return { | |
| 'message': f"'{args['userId']}' already exists." | |
| }, 409 | |
| else: | |
| # create new dataframe containing new values | |
| new_data = pd.DataFrame({ | |
| 'userId': [args['userId']], | |
| 'name': [args['name']], | |
| 'city': [args['city']], | |
| 'locations': [[]] | |
| }) | |
| # add the newly provided values | |
| data = data.append(new_data, ignore_index=True) | |
| data.to_csv('users.csv', index=False) # save back to CSV | |
| return {'data': data.to_dict()}, 200 # return data with 200 OK | |
| def put(self): | |
| parser = reqparse.RequestParser() # initialize | |
| parser.add_argument('userId', required=True) # add args | |
| parser.add_argument('location', required=True) | |
| args = parser.parse_args() # parse arguments to dictionary | |
| # read our CSV | |
| data = pd.read_csv('users.csv') | |
| if args['userId'] in list(data['userId']): | |
| # evaluate strings of lists to lists !!! never put something like this in prod | |
| data['locations'] = data['locations'].apply( | |
| lambda x: ast.literal_eval(x) | |
| ) | |
| # select our user | |
| user_data = data[data['userId'] == args['userId']] | |
| # update user's locations | |
| user_data['locations'] = user_data['locations'].values[0] \ | |
| .append(args['location']) | |
| # save back to CSV | |
| data.to_csv('users.csv', index=False) | |
| # return data and 200 OK | |
| return {'data': data.to_dict()}, 200 | |
| else: | |
| # otherwise the userId does not exist | |
| return { | |
| 'message': f"'{args['userId']}' user not found." | |
| }, 404 | |
| def delete(self): | |
| parser = reqparse.RequestParser() # initialize | |
| parser.add_argument('userId', required=True) # add userId arg | |
| args = parser.parse_args() # parse arguments to dictionary | |
| # read our CSV | |
| data = pd.read_csv('users.csv') | |
| if args['userId'] in list(data['userId']): | |
| # remove data entry matching given userId | |
| data = data[data['userId'] != args['userId']] | |
| # save back to CSV | |
| data.to_csv('users.csv', index=False) | |
| # return data and 200 OK | |
| return {'data': data.to_dict()}, 200 | |
| else: | |
| # otherwise we return 404 because userId does not exist | |
| return { | |
| 'message': f"'{args['userId']}' user not found." | |
| }, 404 | |
| class Locations(Resource): | |
| def get(self): | |
| data = pd.read_csv('locations.csv') # read local CSV | |
| return {'data': data.to_dict()}, 200 # return data dict and 200 OK | |
| def post(self): | |
| parser = reqparse.RequestParser() # initialize parser | |
| parser.add_argument('locationId', required=True, type=int) # add args | |
| parser.add_argument('name', required=True) | |
| parser.add_argument('rating', required=True) | |
| args = parser.parse_args() # parse arguments to dictionary | |
| # read our CSV | |
| data = pd.read_csv('locations.csv') | |
| # check if location already exists | |
| if args['locationId'] in list(data['locationId']): | |
| # if locationId already exists, return 401 unauthorized | |
| return { | |
| 'message': f"'{args['locationId']}' already exists." | |
| }, 409 | |
| else: | |
| # otherwise, we can add the new location record | |
| # create new dataframe containing new values | |
| new_data = pd.DataFrame({ | |
| 'locationId': [args['locationId']], | |
| 'name': [args['name']], | |
| 'rating': [args['rating']] | |
| }) | |
| # add the newly provided values | |
| data = data.append(new_data, ignore_index=True) | |
| data.to_csv('locations.csv', index=False) # save back to CSV | |
| return {'data': data.to_dict()}, 200 # return data with 200 OK | |
| def patch(self): | |
| parser = reqparse.RequestParser() # initialize parser | |
| parser.add_argument('locationId', required=True, type=int) # add args | |
| parser.add_argument('name', store_missing=False) # name/rating are optional | |
| parser.add_argument('rating', store_missing=False) | |
| args = parser.parse_args() # parse arguments to dictionary | |
| # read our CSV | |
| data = pd.read_csv('locations.csv') | |
| # check that the location exists | |
| if args['locationId'] in list(data['locationId']): | |
| # if it exists, we can update it, first we get user row | |
| user_data = data[data['locationId'] == args['locationId']] | |
| # if name has been provided, we update name | |
| if 'name' in args: | |
| user_data['name'] = args['name'] | |
| # if rating has been provided, we update rating | |
| if 'rating' in args: | |
| user_data['rating'] = args['rating'] | |
| # update data | |
| data[data['locationId'] == args['locationId']] = user_data | |
| # now save updated data | |
| data.to_csv('locations.csv', index=False) | |
| # return data and 200 OK | |
| return {'data': data.to_dict()}, 200 | |
| else: | |
| # otherwise we return 404 not found | |
| return { | |
| 'message': f"'{args['locationId']}' location does not exist." | |
| }, 404 | |
| def delete(self): | |
| parser = reqparse.RequestParser() # initialize parser | |
| parser.add_argument('locationId', required=True, type=int) # add locationId arg | |
| args = parser.parse_args() # parse arguments to dictionary | |
| # read our CSV | |
| data = pd.read_csv('locations.csv') | |
| # check that the locationId exists | |
| if args['locationId'] in list(data['locationId']): | |
| # if it exists, we delete it | |
| data = data[data['locationId'] != args['locationId']] | |
| # save the data | |
| data.to_csv('locations.csv', index=False) | |
| # return data and 200 OK | |
| return {'data': data.to_dict()}, 200 | |
| else: | |
| # otherwise we return 404 not found | |
| return { | |
| 'message': f"'{args['locationId']}' location does not exist." | |
| } | |
| api.add_resource(Users, '/users') # add endpoints | |
| api.add_resource(Locations, '/locations') | |
| if __name__ == '__main__': | |
| app.run() # run our Flask app |
In the docs: https://flask-restful.readthedocs.io/en/0.3.8/reqparse.html#basic-arguments it looks like reqparse will be deprecated. Also, there's another argument to add_argument "location="args" that will make sure the add_argument will check the URL string only. After I put in "location="args", parse_args worked. I don't have advice on the deprecation.
Having issues when POST via POSTMAN. GET works fine { "message": "The browser (or proxy) sent a request that this server could not understand." } 127.0.0.1 - - [20/Jan/2023 10:11:31] "POST /users?userId=abc123&name=The%20Rock&city=Los%20Angeles HTTP/1.1" 400 -
thoughts ?
You can lso use parser.add_argument('xxxx', required=True, location="json"). to read JSON values as well, if you instead don't want to have messy query string for every request. Although, if anyone is going to use this beyond a learning example, should probably use something other than reqparse to read the arguments since it will be deprecated.
Having issues when POST via POSTMAN. GET works fine
{
"message": "The browser (or proxy) sent a request that this server could not understand."
}
127.0.0.1 - - [20/Jan/2023 10:11:31] "POST /users?userId=abc123&name=The%20Rock&city=Los%20Angeles HTTP/1.1" 400 -
thoughts ?