Skip to content

Instantly share code, notes, and snippets.

@jamescalam
Last active September 10, 2024 20:13
Show Gist options
  • Save jamescalam/0b309d275999f9df26fa063602753f73 to your computer and use it in GitHub Desktop.
Save jamescalam/0b309d275999f9df26fa063602753f73 to your computer and use it in GitHub Desktop.
A example API using Flask
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
@OldGuyProgrammer
Copy link

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.

@ryanrozanitis
Copy link

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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment