Hello, welcome back to the 4th post of my Flask series, this post would be a continuation of my previous post, please click on any link below to view any post you missed
OUTLINE OF FLASK SERIES
- Structuring a Flask-Restful API for Production
- Creating Custom Error Pages and Handling Exceptions for Flask
- Flask authentication with JWT
- CRUD Operations with Flask (this article)
At the end of this post, you will learn the following
- You will learn how to perform the following basic CRUD (Create, Read, Update, and Delete) operations.
In the last post, we have finished the user registration and login feature, we will work on the diary note features of our application.
First things first, we would update our models.py and create a model for diary note features.
The Note model will be mapped to the notes table in the database and also it would inherit the BaseModel. The fields and methods we defined for our user model are as follows:
Methods
- data: This is used to return the data in a dictionary format.
- get_all_published: This method gets all the published notes.
- get_by_id: This method gets the recipes by ID.
- get_all_draft: This method gets all the unpublished notes.
Fields
- title: title of the notes, the maximum length allowed is 100 characters. It can’t be null.
- notes: contents of the notes.
- publish: This is to indicate whether the user wants to publish it or save it as a draft. It is a Boolean field with a default value of False.
- user_id: this expresses a Foreign relationship with the user and it is not null.
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
db = SQLAlchemy()
class BaseModel(db.Model):
"""Define the base model for all other models."""
__abstract__ = True
id = db.Column(db.Integer(), primary_key=True)
created_on = db.Column(db.DateTime(), server_default=db.func.now(), nullable=False)
updated_on = db.Column(db.DateTime(),nullable=False,
server_default=db.func.now(),
onupdate=db.func.now())
def save(self):
"""Save an instance of the model from the database."""
try:
db.session.add(self)
db.session.commit()
except IntegrityError:
db.session.rollback()
except SQLAlchemyError:
db.session.rollback()
def update(self):
"""Update an instance of the model from the database."""
return db.session.commit()
def delete(self):
"""Delete an instance of the model from the database."""
try:
db.session.delete(self)
db.session.commit()
except SQLAlchemyError:
db.session.rollback()
class User(BaseModel):
__tablename__ = 'user'
username = db.Column(db.String(80), nullable=False, unique=True)
email = db.Column(db.String(200), nullable=False, unique=True)
password = db.Column(db.String(200),nullable=False)
note = db.relationship('Note', backref='user')
@classmethod
def get_by_username(cls, username):
return cls.query.filter_by(username=username).first()
@classmethod
def get_by_email(cls, email):
return cls.query.filter_by(email=email).first()
@classmethod
def get_by_id(cls, id):
return cls.query.filter_by(id=id).first()
class Note(BaseModel):
__tablename__ = 'note'
title = db.Column(db.String(100), nullable=False)
notes = db.Column(db.String(1000))
publish = db.Column(db.Boolean(), default=False)
user_id = db.Column(db.Integer(), db.ForeignKey("user.id"),nullable=False)
def data(self):
return {
'id': self.id,
'title': self.title,
'notes': self.notes,
'user_id': self.user_id
}
@classmethod
def get_all_published(cls):
return cls.query.filter_by(publish=True).all()
@classmethod
def get_all_drafts(cls):
return cls.query.filter_by(publish=False).all()
@classmethod
def get_by_id(cls, note_id):
return cls.query.filter_by(id=note_id).first()
Now make migrations, run the following on your terminal.
flask db migrate -m "migrate notes table" flask db upgrade
Now, please check /migrations/versions/d26c45eccb81_migrate_notes_table under the versions folder. This file is created by Flask-Migrate.
Note that you may get a different revision ID here.
create our view resources.
Create a new file, note.py in the resources folder, and add the following code to it.
from http import HTTPStatus
from flask_jwt_extended import get_jwt_identity, jwt_optional, jwt_required
from flask_restful import Resource
from webargs.fields import Bool, Email, Int, Str
from webargs.flaskparser import use_kwargs
from api.models import Note, User
class NoteListResource(Resource):
@jwt_required
def get(self, user_notes=None):
current_user = get_jwt_identity()
user_notes = Note.query.filter_by(user_id=current_user).first()
notes = user_notes.get_all_published()
data = []
if notes:
for note in notes:
data.append(note.data())
return {"data": data}, HTTPStatus.OK
return {"msg": "no notes available"}, HTTPStatus.BAD_REQUEST
@use_kwargs(
{
"title": Str(required=True, location="json"),
"notes": Str(required=True, location="json"),
}
)
@jwt_required
def post(self, title, notes):
current_user = get_jwt_identity()
note = Note(title=title, notes=notes, user_id=current_user)
saved_notes = note.save()
return (
{
"msg": "successfully created notes",
"data": {
"title_of_post": title,
"contents_of_post": notes,
"user_id": current_user,
},
},
HTTPStatus.CREATED,
)
class NoteResource(Resource):
@use_kwargs(
{
"note_id": Int(location="path"),
"title": Str(required=True, location="json"),
"notes": Str(required=True, location="json"),
}
)
@jwt_required
def put(self, note_id, title, notes):
note = Note.get_by_id(note_id=note_id)
if note is None:
return {"message": "Note not found"}, HTTPStatus.NOT_FOUND
current_user = get_jwt_identity()
if current_user != note.user_id:
return {"message": "Access is not allowed"}, HTTPStatus.FORBIDDEN
note.title = title
note.notes = notes
return (
{"msg": "records updated successfully", "data": note.data()},
), HTTPStatus.OK
@use_kwargs({"note_id": Int(location="path")})
@jwt_required
def delete(self, note_id):
note = Note.get_by_id(note_id=note_id)
if note is None:
return {"message": "Note not found"}, HTTPStatus.NOT_FOUND
current_user = get_jwt_identity()
if current_user != note.user_id:
return {"message": "Access is not allowed"}, HTTPStatus.FORBIDDEN
note.delete()
return {"msg": "action completed, note has been deleted"}, HTTPStatus.OK
class NotePublishResource(Resource):
@use_kwargs({"note_id": Int(location="path")})
@jwt_required
def put(self, note_id):
note = Note.get_by_id(note_id=note_id)
if note is None:
return {"message": "Note not found"}, HTTPStatus.NOT_FOUND
current_user = get_jwt_identity()
if current_user != note.user_id:
return {"message": "Access is not allowed"}, HTTPStatus.FORBIDDEN
note.publish = True
note.save()
return {"msg": "your note has been published succesfully"}, HTTPStatus.OK
class DraftNoteListResource(Resource):
@jwt_required
def get(self, user_notes=None):
current_user = get_jwt_identity()
user_notes = Note.query.filter_by(user_id=current_user).first()
notes = user_notes.get_all_drafts()
data = []
if notes:
for note in notes:
data.append(note.data())
return {"data": data}, HTTPStatus.OK
return {"msg": "no notes available"}, HTTPStatus.BAD_REQUEST
From the above code snippets, we created four endpoints.
NoteListResource: This resource has 2 functions
- Getting published notes (get request) and it can’t be accessed without a token, that is what the @jwt_required decorator enforces. It checks the current user and filters out published notes written by the logged-in user.
- Creating a new diary notes (post request), the @use_kwargs decorator
@use_kwargs({ "title":Str(required=True,location="ddjson"), "notes":Str(required=True, location="json")})handles the client request and will search for arguments from the request body as JSON, we specified the location as JSON from which to load data from.
NoteResource: This resource has 2 functions
- Update a note with a specific id (put request), we passed the
@use_kwargs({"note_id": Int(location=" path")})and its location is a path, so we can only edit and update notes with that id passed in the URL. This resource or endpoint can’t be accessed without a token, that is what the @jwt_required decorator enforces. It checks the current user and filters out published notes written by the logged-in user. - Delete a note (either draft or published) with a specific id (delete request), we passed the
@use_kwargs({"note_id": Int(location=" path")})and its location is a path, so we can delete notes with that id passed in the URL. This resource or endpoint can’t be accessed without a token, that is what the @jwt_required decorator enforces. It checks the current user and filters out published notes written by the logged-in use
NotePublishResource: This resource is an endpoint for publishing your notes with a specific id (get request), we passed the@use_kwargs({"note_id": Int(location=" path")}) and its location is a path, so we can publish notes with that id passed in the URL.
DraftNoteListResource: This resource is an endpoint for getting all your drafts that are not yet published.
Resourceful Routing
We are going to route our resources and pass our URLs to the add_resource() method.
Update your app.py and add the following code.
from flasgger import Swagger
from flask import Flask
from flask_jwt_extended import JWTManager
from flask_migrate import Migrate
from flask_restful import Api
from webargs.flaskparser import abort, parser
from werkzeug import exceptions
from api.config import env_config
from api.models import db
from resources.default import DefaultResource
from resources.notes import (DraftNoteListResource, NoteListResource,
NotePublishResource, NoteResource)
from resources.user import (RefreshAccessTokenResource,
RevokeAccessTokenResource, UserInfoResource,
UserLoginResource, UserRegistrationResource,
black_list)
from utils import errors
api = Api()
jwt = JWTManager()
def create_app(config_name):
import resources
app = Flask(__name__)
app.config.from_object(env_config[config_name])
db.init_app(app) # Add db session, new code here
Migrate(app, db) # new code here
# register api
api.init_app(app)
Swagger(app)
jwt.init_app(app)
# error handling
@jwt.token_in_blacklist_loader
def check_if_token_in_blacklist(decrypted_token):
jti = decrypted_token["jti"]
return jti in black_list
app.register_error_handler(exceptions.NotFound, errors.handle_404_errors)
app.register_error_handler(
exceptions.InternalServerError, errors.handle_server_errors
)
app.register_error_handler(exceptions.BadRequest, errors.handle_400_errors)
app.register_error_handler(FileNotFoundError, errors.handle_400_errors)
app.register_error_handler(TypeError, errors.handle_400_errors)
app.register_error_handler(KeyError, errors.handle_404_errors)
app.register_error_handler(AttributeError, errors.handle_400_errors)
app.register_error_handler(ValueError, errors.handle_400_errors)
app.register_error_handler(AssertionError, errors.handle_400_errors)
# new code
@parser.error_handler
def handle_request_parsing_error(
err, req, schema, *, error_status_code, error_headers
):
"""webargs error handler that uses Flask-RESTful's abort function to return
a JSON error response to the client.
"""
abort(error_status_code, errors=err.messages)
return app
# register our urls for user module
api.add_resource(
UserRegistrationResource, "/v1/user/register/", endpoint="user_registration"
)
api.add_resource(UserLoginResource, "/v1/user/login/", endpoint="user_login")
api.add_resource(UserInfoResource, "/v1/user/user_info/", endpoint="user_info")
api.add_resource(
RefreshAccessTokenResource, "/v1/user/refresh_token/", endpoint="refresh_token"
)
api.add_resource(
RevokeAccessTokenResource, "/v1/user/signout_access/", endpoint="signout_access"
)
# register our urls for note module
api.add_resource(NoteListResource, "/v1/notes/", endpoint="notes")
api.add_resource(NoteResource, "/v1/notes/<int:note_id>", endpoint="note_id")
api.add_resource(
NotePublishResource, "/v1/publish_note/<int:note_id>", endpoint="publish_note"
)
api.add_resource(DraftNoteListResource, "/v1/notes/draft/", endpoint="draft")
# register url for default
api.add_resource(DefaultResource, "/", endpoint="home")
testing Web Services with HTTPie
CREATE A NEW NOTE
Let’s try creating notes without logging in as an authenticated user.
Run the command on your terminal
http POST http://127.0.0.1:5000/v1/notes/ title="First day in Lagos" notes="A visit to Yaba"
Our action was unsuccessful because it requires an access token.
Now let’s login as a user and create our notes, log in an existing user, and copy the access token to create a new note.
http POST :5000/v1/user/login/ email="oluch@gmail.com" password="445hD#@%yu78" http POST http://127.0.0.1:5000/v1/notes/ title="First day in Lagos" notes="A visit to Yaba" "Authorization:Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE1OTYyMTg1OTEsIm5iZiI6MTU5NjIxODU5MSwianRpIjoiOTFlYmZjMGMtNGQ3ZC00ZmU3LWExNGUtN2JhZWJkYWQwZWYxIiwiZXhwIjoxNTk2MjMyOTkxLCJpZGVudGl0eSI6MSwiZnJlc2giOnRydWUsInR5cGUiOiJhY2Nlc3MifQ.bFKHeNOMYDbhZJALLHsDfP3Tpe9qDzNYWDTEtlQD8Ic"
You can see our action was successful, The HTTP status code 201 here means the note was created successfully.
PUBLISH A NOTE
Let us publish note with id 1
http PUT http://127.0.0.1:5000/v1/publish_note/1 "Authorization:Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE1OTYyMTg1OTEsIm5iZiI6MTU5NjIxODU5MSwianRpIjoiOTFlYmZjMGMtNGQ3ZC00ZmU3LWExNGUtN2JhZWJkYWQwZWYxIiwiZXhwIjoxNTk2MjMyOTkxLCJpZGVudGl0eSI6MSwiZnJlc2giOnRydWUsInR5cGUiOiJhY2Nlc3MifQ.bFKHeNOMYDbhZJALLHsDfP3Tpe9qDzNYWDTEtlQD8Ic"
get all published post
http GET http://127.0.0.1:5000/v1/notes/ "Authorization:Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE1OTYyMTg1OTEsIm5iZiI6MTU5NjIxODU5MSwianRpIjoiOTFlYmZjMGMtNGQ3ZC00ZmU3LWExNGUtN2JhZWJkYWQwZWYxIiwiZXhwIjoxNTk2MjMyOTkxLCJpZGVudGl0eSI6MSwiZnJlc2giOnRydWUsInR5cGUiOiJhY2Nlc3MifQ.bFKHeNOMYDbhZJALLHsDfP3Tpe9qDzNYWDTEtlQD8Ic"
UPDATE A NOTE
Update post with id 1
http PUT http://127.0.0.1:5000/v1/notes/1 title="First day in South Africa" notes="I love Capetown" "Authorization:Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE1OTYyMTg1OTEsIm5iZiI6MTU5NjIxODU5MSwianRpIjoiOTFlYmZjMGMtNGQ3ZC00ZmU3LWExNGUtN2JhZWJkYWQwZWYxIiwiZXhwIjoxNTk2MjMyOTkxLCJpZGVudGl0eSI6MSwiZnJlc2giOnRydWUsInR5cGUiOiJhY2Nlc3MifQ.bFKHeNOMYDbhZJALLHsDfP3Tpe9qDzNYWDTEtlQD8Ic"
DELETE A NOTE
Delete note with id 3
http DELETE http://127.0.0.1:5000/v1/notes/3 "Authorization:Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE1OTYyOTA0MDgsIm5iZiI6MTU5NjI5MDQwOCwianRpIjoiMzZjNDE4MDktNmRiNy00Y2NhLTk3MmQtMTBkOWUxMDczOGZkIiwiZXhwIjoxNTk2MzA0ODA4LCJpZGVudGl0eSI6MSwiZnJlc2giOnRydWUsInR5cGUiOiJhY2Nlc3MifQ.c3_Sj_WcWr6aitfemtmNX-n_8r77SWUmozyj_VKpRUM"
get all drafts (unpublished NOTES)
http GET http://127.0.0.1:5000/v1/notes/draft/ "Authorization:Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE1OTYyOTA0MDgsIm5iZiI6MTU5NjI5MDQwOCwianRpIjoiMzZjNDE4MDktNmRiNy00Y2NhLTk3MmQtMTBkOWUxMDczOGZkIiwiZXhwIjoxNTk2MzA0ODA4LCJpZGVudGl0eSI6MSwiZnJlc2giOnRydWUsInR5cGUiOiJhY2Nlc3MifQ.c3_Sj_WcWr6aitfemtmNX-n_8r77SWUmozyj_VKpRUM"
We are done with our 4th post on my Flask series, next post I would be implementing Object Serialization with marshmallow.
The link to Github tutorial is here
Until then happy coding.
