Home » Tutorials » How to Create a Robust RESTful API with Flask in Python

How to Create a Robust RESTful API with Flask in Python

By /

Creating a RESTful API is a fundamental skill for any modern web developer. It allows different applications to communicate with each other over the web, enabling the integration of various services and systems. RESTful APIs are widely used due to their simplicity, scalability, and stateless nature, making them an ideal choice for building web services that can grow with your application.

In this tutorial, we’ll dive into how to create a robust RESTful API with Flask in Python. You’ll learn how to configure your Flask application for different environments, set up a database to manage data efficiently, and implement secure authentication to protect your routes. We will also cover how to create and test RESTful API endpoints using Pytest, ensuring your application runs smoothly and reliably. So, let’s get started!

Table of Contents

Getting Started

First, you need to organize your Python scripts to ensure your Flask application works correctly. Create two folders within your Flask project: app and tests, each containing the relevant Python scripts:

Necessary Libraries

Well, before we dive into the code, make sure to install these libraries using your terminal or command prompt by running the following commands:

$ pip install Flask
$ pip install Flask-RESTful
$ pip install Flask-SQLAlchemy
$ pip install Flask-HTTPAuth
$ pip install pytest

Configuration and Setting the Stage

The first step in creating our application is to configure it to be adaptable for any environment, whether it is development or production. This is the goal of this script. Now, with that being said, let’s dissect this script and see how it works:

Imports:

To begin with, we need to access the environment variables so that our application can adapt. For this, we use the os module, as it allows us to interact with the operating system.

import os

Base Configuration Class:

This is the blueprint that contains the common settings for all environments. How does it work? You could say that this class knows how to find its database. How does it find it, you might ask? Well, it uses os.getenv() to check if the environment variables have a database URL. If a URL is found, it uses the SQLAlchemy database; if not, it defaults to the SQLite database. To save memory, it keeps everything lean by disabling the tracking of modifications to objects, setting it to False.

class Config:
   SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL', 'sqlite:///tasklist.db')
   SQLALCHEMY_TRACK_MODIFICATIONS = False

Development Configuration:

Here we have the development subclass of the config class. Why did we add this subclass? Well, the simple answer is to make our work easier by adding certain settings to the config class that are specific to the development environment.

What are those settings? The first one is DEBUG, which provides detailed error messages and automatically reloads the server on code changes. The second and last one is SECRET_KEY, which is fetched from the environment variable for the application if it exists; otherwise, we use the dev_key.

class DevelopmentConfig(Config):
   DEBUG = True
   SECRET_KEY = os.getenv('SECRET_KEY', 'dev_key')

Production Configuration:

Now, here is the production part. This means we have finished the development of our app, so we no longer need the DEBUG setting and set it to False. Since we are now in the production period, we need a new SECRET_KEY that is specific to production, fetched from the environment variable. If there’s no SECRET_KEY in the variable, it is substituted with the prod_key.

class ProductionConfig(Config):
   DEBUG = False
   SECRET_KEY = os.getenv('SECRET_KEY', 'prod_key')

config.py

import os

class Config:
   SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL', 'sqlite:///tasklist.db')
   SQLALCHEMY_TRACK_MODIFICATIONS = False


class DevelopmentConfig(Config):
   DEBUG = True
   SECRET_KEY = os.getenv('SECRET_KEY', 'dev_key')


class ProductionConfig(Config):
   DEBUG = False
   SECRET_KEY = os.getenv('SECRET_KEY', 'prod_key')

Setting Up the Database

After we finish configuring our application, it’s time to set up the database, which is the heartbeat of our application as it stores and manages all the vital data. Now, let’s go through this script piece by piece:

Imports:

Well, the first thing we need is a toolkit that makes working with databases in Flask a breeze, which is why we import SQLAlchemy. Since we’re going to store tasks in the database, we need a timekeeper to track when tasks are created and updated, which is why we import datetime.

from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

Initializing the Database:

Now it’s time to create an instance of SQLAlchemy to interact with the database and store and retrieve tasks.

db = SQLAlchemy()

Defining the Task Model:

For this step, we have the Task class, whose objective is to define how tasks are structured in the database. Now, let’s see how it structures them:

  • First, by giving a unique identifier (id) to each task so they can be tracked.
  • Second, by providing a detailed description of the task, which serves as a mandatory briefing for every mission.
  • Third, by recording the time at which the task was created.Fourth, by timestamping whenever the task is updated.
  • Fifth, by using a boolean flag to indicate whether the task is deleted or not, meaning the task can be archived but never lost.
class Task(db.Model):
   id = db.Column(db.Integer, primary_key=True)
   description = db.Column(db.String(255), nullable=False)
   created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
   updated_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow)
   deleted = db.Column(db.Boolean, nullable=False, default=False)

Mission Brief:

This part provides a quick status check, offering a perfect summary of each task (id and description).

   def __repr__(self):
       return f'<Task {self.id}: {self.description}>'

database.py

from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

db = SQLAlchemy()

class Task(db.Model):
   id = db.Column(db.Integer, primary_key=True)
   description = db.Column(db.String(255), nullable=False)
   created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
   updated_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow)
   deleted = db.Column(db.Boolean, nullable=False, default=False)

   def __repr__(self):
       return f'<Task {self.id}: {self.description}>'

Authentication

What do you think comes after configuring the application and setting up its database? The answer is authentication. Just as you wouldn’t want anyone to access your phone, you don’t want anyone to access the tasks you store. The objective of this script is to grant access only to authorized users.

Imports:

Our only import is the HTTPBasicAuth module, which allows us to perform basic HTTP authentication in our Flask application.

from flask_httpauth import HTTPBasicAuth

Initializing the Authentication:

Here, we activate our security system by initializing an instance of HTTPBasicAuth to manage authentication.

auth = HTTPBasicAuth()

Defining User Credentials:

In this section, we define a dictionary named users where the keys are usernames and the values are their passwords.

users = {
   "admin": "password123",
   "user": "password"
}

Password Retrieval Function:

The objective of this function is to retrieve the password using the @auth.get_password decorator. It verifies and recognizes if the username is among the list of users; if the username is not found in the list, access is denied.

@auth.get_password
def get_password(username):
   if username in users:
       return users.get(username)
   return None

auth.py

from flask_httpauth import HTTPBasicAuth

auth = HTTPBasicAuth()

users = {
   "admin": "password123",
   "user": "password"
}

@auth.get_password
def get_password(username):
   if username in users:
       return users.get(username)
   return None

RESTful API Resources

Now that we have completed setting up authentication, it is time to carry out the orders (API requests). This script sets up powerful RESTful API resources that handle everything from getting tasks to creating, updating, and deleting them. Let’s see how this script actually works:

Imports:

Just like the previous scripts, we need to import the necessary tools:

  • Our first import is the Resource class to create a RESTful API endpoint from Flask-RESTful.
  • Then we include request and jsonify to handle incoming data and return responses.
  • Next, we bring in Task and db to act as our database operatives.
  • Finally, auth to act as a security guard, allowing access only to authorized users.
from flask_restful import Resource
from flask import request, jsonify
from app.database import Task, db
from app.auth import auth

Task Resource:

The objective of this class is to manage our collection of tasks through:

  • A GET request to retrieve all non-deleted tasks.
  • A POST request to create a new task and add it to our mission dossier.
class TaskResource(Resource):
   decorators = [auth.login_required]


   def get(self):
       try:
           tasks = Task.query.filter_by(deleted=False).all()
           task_data = [{'task_id': task.id, 'task_description': task.description} for task in tasks]
           return {'tasks': task_data}, 200
       except Exception as e:
           return {'message': str(e)}, 500


   def post(self):
       try:
           task_details = request.get_json()
           if not task_details or not task_details.get('description'):
               return {'message': 'Task description is required'}, 400


           new_task = Task(description=task_details['description'])
           db.session.add(new_task)
           db.session.commit()
           return {'message': 'Task created',
                   'task': {'task_id': new_task.id, 'task_description': new_task.description}}, 201
       except Exception as e:
           return {'message': str(e)}, 500

Task Detail Resource:

If the previous class handles a collection of tasks, then this one handles individual tasks identified by task_id. What do I mean by handling? Well, it’s either through:

  • A PUT request that updates the details of a specific task.
  • A DELETE request that deactivates a task by marking it as deleted.
class TaskDetailResource(Resource):
   decorators = [auth.login_required]


   def put(self, task_id):
       try:
           task = Task.query.get_or_404(task_id)
           task_details = request.get_json()


           if not task_details or not task_details.get('description'):
               return {'message': 'Task description is required'}, 400


           task.description = task_details['description']
           db.session.commit()
           return {'message': 'Task updated', 'task': {'task_id': task.id, 'task_description': task.description}}, 200
       except Exception as e:
           return {'message': str(e)}, 500


   def delete(self, task_id):
       try:
           task = Task.query.get_or_404(task_id)
           task.deleted = True
           db.session.commit()
           return {'message': 'Task deleted'}, 200
       except Exception as e:
           return {'message': str(e)}, 500

task_resources.py

from flask_restful import Resource
from flask import request, jsonify
from app.database import Task, db
from app.auth import auth




class TaskResource(Resource):
   decorators = [auth.login_required]


   def get(self):
       try:
           tasks = Task.query.filter_by(deleted=False).all()
           task_data = [{'task_id': task.id, 'task_description': task.description} for task in tasks]
           return {'tasks': task_data}, 200
       except Exception as e:
           return {'message': str(e)}, 500


   def post(self):
       try:
           task_details = request.get_json()
           if not task_details or not task_details.get('description'):
               return {'message': 'Task description is required'}, 400


           new_task = Task(description=task_details['description'])
           db.session.add(new_task)
           db.session.commit()
           return {'message': 'Task created',
                   'task': {'task_id': new_task.id, 'task_description': new_task.description}}, 201
       except Exception as e:
           return {'message': str(e)}, 500


class TaskDetailResource(Resource):
   decorators = [auth.login_required]


   def put(self, task_id):
       try:
           task = Task.query.get_or_404(task_id)
           task_details = request.get_json()


           if not task_details or not task_details.get('description'):
               return {'message': 'Task description is required'}, 400


           task.description = task_details['description']
           db.session.commit()
           return {'message': 'Task updated', 'task': {'task_id': task.id, 'task_description': task.description}}, 200
       except Exception as e:
           return {'message': str(e)}, 500


   def delete(self, task_id):
       try:
           task = Task.query.get_or_404(task_id)
           task.deleted = True
           db.session.commit()
           return {'message': 'Task deleted'}, 200
       except Exception as e:
           return {'message': str(e)}, 500

Assembling the App Factory

This is the part where we assemble everything by configuring our application to use the database and RESTful API resources, preparing it for launch.

Imports:

First, we import Flask to create our Flask application. Then, we import Api to create a RESTful API. Next, we import db, the SQLAlchemy database instance, from our app.

After that, we import the DevelopmentConfig subclass from the config file, and the TaskResource as well as the TaskDetailResource classes from the task_resources module we made earlier.

import logging
from flask import Flask
from flask_restful import Api
from app.database import db
from config import DevelopmentConfig
from app.task_resources import TaskResource, TaskDetailResource
from app.auth import auth

Initialize Flask Application:

Here, we create a new Flask application instance and then use the config class from the config file to configure it.

def create_app(config_class=DevelopmentConfig):
   app = Flask(__name__)
   app.config.from_object(config_class)

Initialize the Database:

Next, we allow the app to connect with the database by initializing the database with the app context.

   db.init_app(app)

Set Up the API:

For this step, we create an API instance and link it to the Flask application. Then, we add the task resources to it with their respective endpoints:

  • TaskResource for /api/tasks
  • TaskDetailResource for /api/tasks/<int:task_id>
   api = Api(app)
   api.add_resource(TaskResource, '/api/tasks')
   api.add_resource(TaskDetailResource, '/api/tasks/<int:task_id>')


   # Setup logging
   logging.basicConfig(filename='app.log', level=logging.DEBUG,
                       format='%(asctime)s %(levelname)s %(name)s %(threadName)s : %(message)s')
   app.logger.info('App startup')

Return the App:

Now once the flask application is configured it is returned.

   return app

_init_.py

import logging
from flask import Flask
from flask_restful import Api
from app.database import db
from config import DevelopmentConfig
from app.task_resources import TaskResource, TaskDetailResource
from app.auth import auth


def create_app(config_class=DevelopmentConfig):
   app = Flask(__name__)
   app.config.from_object(config_class)


   db.init_app(app)


   api = Api(app)
   api.add_resource(TaskResource, '/api/tasks')
   api.add_resource(TaskDetailResource, '/api/tasks/<int:task_id>')


   # Setup logging
   logging.basicConfig(filename='app.log', level=logging.DEBUG,
                       format='%(asctime)s %(levelname)s %(name)s %(threadName)s : %(message)s')
   app.logger.info('App startup')

   return app

Running the Flask Application

Now that our application is ready, it’s time to bring it to life and run it. But before that:

Imports:

We start by importing the create_app function that configures our application. Then, we import the SQLAlchemy database instance.

from app import create_app
from app.database import db

Running the Application:

This part ensures that this script only runs when executed directly and not when imported as a module. It also triggers create_app to assemble all components into a cohesive unit, ensures that all necessary database tables are created, and finally starts the application, specifically the Flask development server.

if __name__ == '__main__':
   app = create_app()
   with app.app_context():
       db.create_all()
   app.run()

main.py

from app import create_app
from app.database import db


if __name__ == '__main__':
   app = create_app()
   with app.app_context():
       db.create_all()
   app.run()

Testing: the Mission Debrief

This is where we test our Flask application by verifying every function to ensure they work perfectly.

Imports:

To test our App we import pytest, our powerful ally and testing framework. Then, we bring the create_app function to set up and configure our Flask application. Finally, we import the database and Task model.

import pytest
from app import create_app
from app.database import db, Task

App Fixture:

This section uses an SQLite database for tests, creates the database tables before tests, and cleans up afterward.

@pytest.fixture
def app():
   app = create_app()
   app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test_tasklist.db'
   app.config['TESTING'] = True


   with app.app_context():
       db.create_all()
       yield app
       db.drop_all()

Client Fixture:

This simulates HTTP requests to make it seem like there’s a client interacting with the server.

@pytest.fixture
def client(app):
   return app.test_client()

Testing Retrieving Tasks:

This one simulates a GET request to fetch tasks while ensuring that the response is successful and returns an empty list.

def test_get_tasks(client):
   response = client.get('/api/tasks', auth=('admin', 'password123'))
   assert response.status_code == 200
   assert response.get_json() == {'tasks': []}

Testing creating a Task:

Now, instead of GET we will simulate the POST request to see if the task is created successfully.

def test_create_task(client):
   response = client.post('/api/tasks', json={'description': 'Test Task'}, auth=('admin', 'password123'))
   assert response.status_code == 201
   data = response.get_json()
   assert data['message'] == 'Task created'
   assert data['task']['task_description'] == 'Test Task'

Testing Updating a Task:

This time we will simulate the PUT request to see if the task details are updated successfully.

def test_update_task(client):
   response = client.post('/api/tasks', json={'description': 'Update Task'}, auth=('admin', 'password123'))
   task_id = response.get_json()['task']['task_id']
   response = client.put(f'/api/tasks/{task_id}', json={'description': 'Updated Task'}, auth=('admin', 'password123'))
   assert response.status_code == 200
   data = response.get_json()
   assert data['message'] == 'Task updated'
   assert data['task']['task_description'] == 'Updated Task'

Testing Deleting a Task:

Here we simulate the delete request that marks the task as deleted, then see if the task is marked as deleted as it should be.

def test_delete_task(client):
   response = client.post('/api/tasks', json={'description': 'Delete Task'}, auth=('admin', 'password123'))
   task_id = response.get_json()['task']['task_id']
   response = client.delete(f'/api/tasks/{task_id}', auth=('admin', 'password123'))
   assert response.status_code == 200
   assert response.get_json()['message'] == 'Task deleted'

test_app.py

import pytest
from app import create_app
from app.database import db, Task


@pytest.fixture
def app():
   app = create_app()
   app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test_tasklist.db'
   app.config['TESTING'] = True


   with app.app_context():
       db.create_all()
       yield app
       db.drop_all()


@pytest.fixture
def client(app):
   return app.test_client()


def test_get_tasks(client):
   response = client.get('/api/tasks', auth=('admin', 'password123'))
   assert response.status_code == 200
   assert response.get_json() == {'tasks': []}


def test_create_task(client):
   response = client.post('/api/tasks', json={'description': 'Test Task'}, auth=('admin', 'password123'))
   assert response.status_code == 201
   data = response.get_json()
   assert data['message'] == 'Task created'
   assert data['task']['task_description'] == 'Test Task'


def test_update_task(client):
   response = client.post('/api/tasks', json={'description': 'Update Task'}, auth=('admin', 'password123'))
   task_id = response.get_json()['task']['task_id']
   response = client.put(f'/api/tasks/{task_id}', json={'description': 'Updated Task'}, auth=('admin', 'password123'))
   assert response.status_code == 200
   data = response.get_json()
   assert data['message'] == 'Task updated'
   assert data['task']['task_description'] == 'Updated Task'


def test_delete_task(client):
   response = client.post('/api/tasks', json={'description': 'Delete Task'}, auth=('admin', 'password123'))
   task_id = response.get_json()['task']['task_id']
   response = client.delete(f'/api/tasks/{task_id}', auth=('admin', 'password123'))
   assert response.status_code == 200
   assert response.get_json()['message'] == 'Task deleted'

Example

First, we use cd to navigate to our Flask project directory. For example:

cd C:\Users\gg\PycharmProjects\FlaskProject

Second, we start our Flask app:

python main.py

Third, we create a new task with the following command. You can add any description and name it however you like:

curl -u admin:password123 -X POST -H "Content-Type: application/json" -d "{\"description\": \"New Task\"}" http://127.0.0.1:5000/api/tasks

Fourth, we can update a task like this:

curl -u admin:password123 -X PUT -H "Content-Type: application/json" -d "{\"description\": \"Updated Task\"}" http://127.0.0.1:5000/api/tasks/2

We can see all our tasks with this command:

curl -u admin:password123 http://127.0.0.1:5000/api/tasks

We can delete a task like this (choose the number of the task you want to delete; in this case, it is task number one):

curl -u admin:password123 -X DELETE http://127.0.0.1:5000/api/tasks/1

Happy Coding!

Subscribe for Top Free Python Tutorials!

Receive the best directly.  Elevate Your Coding Journey!

Read Also:

How to Render HTML Templates in Flask How to Translate Languages with Flask in Python How to Build a Weather App with Flask in Python How to Make a YouTube Video Downloader with Flask in Python

Leave a Comment

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

Scroll to Top
×