CRUD Operations Using Flask

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

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"

flask login

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"

crud

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"

CRUD

get all published post
http GET http://127.0.0.1:5000/v1/notes/   "Authorization:Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE1OTYyMTg1OTEsIm5iZiI6MTU5NjIxODU5MSwianRpIjoiOTFlYmZjMGMtNGQ3ZC00ZmU3LWExNGUtN2JhZWJkYWQwZWYxIiwiZXhwIjoxNTk2MjMyOTkxLCJpZGVudGl0eSI6MSwiZnJlc2giOnRydWUsInR5cGUiOiJhY2Nlc3MifQ.bFKHeNOMYDbhZJALLHsDfP3Tpe9qDzNYWDTEtlQD8Ic"

crud

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"

crud

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.

 

 

 

Leave a Reply

Your email address will not be published. Required fields are marked *