Archive for February 2025

How To Unit Test a Flask Restful Get Endpoint With A Parameter Using Python

In this article, we are going to explore how to unit test a simple restful endpoint using Python 3 and the modules Flask, Flask_restx, and UnitTest. The test will focus on testing an endpoint that has a parameter for specifying the name of the required setting.

Creating The Application

First, we are going to create a simple flask application, the application will host a GET endpoint called /Setting/{sname} that returns the value for a single setting.

  • Create a text file called endPointGet.py and add the following code.
from flask import Flask
from flask_restx import Resource, Api
from flask_cors import CORS

app = Flask(__name__)
CORS(app)  # Allow CORS for all routes
api = Api(app, version='1.0', title='EndPoint Get', description='Example Restful Service')

mySettingsData = [
            {'Name': 'Setting1', 'Value': 'Value1'},
            {'Name': 'Setting2', 'Value': 'Value2'}
        ]

def getSettingValue(sname):
    for setting in mySettingsData:
        if setting['Name'] == sname:
            return setting['Value']
    return None

@api.route('/Settings', doc={"description": "Get all settings"})
class GetSettings(Resource):
    def get(self):
        return mySettingsData

@api.route('/Setting/<string:sname>', doc={"description": "Get one setting value"})
@api.param('sname', 'Setting Name')
class GetSetting(Resource):
    def get(self, sname):
        value = getSettingValue(sname)
        if value == None:
            return {'message': f'Invalid setting name. {sname} does not exist'}, 400
        else:
            return value

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5500, debug=False)

Running The Application

To run the application:

  • From the terminal execute the python file.
    python endPointGet.py

1

  • Open a web browser and navigate to http://127.0.0.1:5500, the page will show the swagger document for the Flask application.
    2

  • Click default to expand the list of endpoints
    3

  • Click the GET button next to /Setting/{sname}.
    4

  • Click the Try it out button.
    5

  • Enter the name of the setting into sname and then click the Execute button.
    6

  • The value of the setting will be displayed in the Response Body
    7

  • Entering a setting name that does not exist will result in an error code 400.
    8

Unit Testing The Application

The unit test will test two scenarios when the setting name exists and when the setting name does not exist. To unit test the flask application:

  • Create a text file called endPointGetTests.py and add the following code
import unittest
import json
from endPointGet import *

class SystemSettingsTestCase(unittest.TestCase):
    def setUp(self):
        """Set up for unit tests, function is executed before tests begin"""

        self.client = app.test_client()
        app.config['TESTING'] = True

    def test_get_setting(self):
        """Test confirms that GET /Setting/{sname} will return a setting value"""

        # Arrange
        expected_result = 'Value1'

        # Act
        response = self.client.get('/Setting/Setting1') # Make a GET request to the /Setting/{sname} endpoint

        # Assert
        self.assertEqual(response.status_code, 200) # Check that the response status code is 200 OK
        actual_result = json.loads(response.data) # Parse the JSON response
        self.assertEqual(actual_result, expected_result) # Check that the response contains the expected data

    def test_get_setting_bad_setting_name(self):
        """Test confirms that GET /Setting/{sname} will return 400 when the setting name does not exisit"""

        # Arrange
        sname = 'Bad_Setting_Name'
        expected_message = f'Invalid setting name. {sname} does not exist'

        # Act
        response = self.client.get(f'/Setting/{sname}') # Make a GET request to the /Settings endpoint

        # Assert
        self.assertEqual(response.status_code, 400) # Check that the response status code is 400 BAD
        self.assertEqual(response.status, '400 BAD REQUEST') # Check the response status message        
        response_json = response.get_json() # Extract the JSON data from the response    
        self.assertIn('message', response_json) # Ensure the 'message' key is in the response
        self.assertEqual(response_json['message'], expected_message) # Check that the response message contains the expected message

if __name__ == '__main__':
    unittest.main()
  • To run the test, execute the Python file endPointGetTests.py from the terminal.

python endPointGetTests.py

  • The log messages from a successful test will end with OK.
    9

How To Unit Test a Flask Restful Get Endpoint Using Python

In this article, we are going to explore how to unit test a simple GET restful endpoint using Python 3 and the modules Flask, Flask_restx, and UnitTest.

Creating The Application

First, we are going to create a simple flask application, the application will host a GET endpoint called /Settings that returns a list of setting names and values in JSON format.

  • Create a text file called endPointGet.py and add the following code.
from flask import Flask
from flask_restx import Namespace, Resource, Api, fields, reqparse
from flask_cors import CORS

app = Flask(__name__)
CORS(app)  # Allow CORS for all routes
api = Api(app, version='1.0', title='EndPoint Get', description='Example Restful Service')

mySettingsData = [
            {'Name': 'Setting1', 'Value': 'Value1'},
            {'Name': 'Setting2', 'Value': 'Value2'}
        ]

@api.route('/Settings', doc={"description": "Get all settings"})
class GetSettings(Resource):
    def get(self):
        return mySettingsData

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5500, debug=False)

Running The Application

To run the application:

  • From the terminal execute the python file.
    python endPointGet.py

1

  • Open a web browser and navigate to http://127.0.0.1:5500, the page will show the swagger document for the Flask application.
    2

  • Click default to expand the list of endpoints

  • Click the GET button next to /Settings.
    3

  • Click the Try it out button.
    4

  • Click the Execute button.
    5

  • The response to the request will be displayed in the response body section.
    6

Unit Testing The Application

To unit test the flask application:

  • Create a text file called endPointGetTests.py and add the following code
import unittest
import json
from endPointGet import *

class SystemSettingsTestCase(unittest.TestCase):
    def setUp(self):
        """Set up for unit tests, function is executed before tests begin"""

        self.client = app.test_client()
        app.config['TESTING'] = True

    def test_get_settings(self):
        """Test confirms that GET /Settings will return a list of all the system settings"""

        # Arrange 
        expected_result = [
            {'Name': 'Setting1', 'Value': 'Value1'},
            {'Name': 'Setting2', 'Value': 'Value2'}
        ]

        # Act
        response = self.client.get('/Settings') # Make a GET request to the /Settings endpoint

        # Assert
        self.assertEqual(response.status_code, 200) # Check that the response status code is 200 OK
        actual_result = json.loads(response.data) # Parse the JSON response        
        self.assertEqual(actual_result, expected_result) # Check that the response contains the expected data

if __name__ == '__main__':
    unittest.main()
  • To run the test, execute the Python file endPointGetTests.py from the terminal.

python endPointGetTests.py

  • The log messages from a successful test will end with OK.
    7

  • The log messages from a failed test will end with the number of failures.
    8

How To Access Sqlite3 Database Using Python

In this article, we are going to create a Python class that can read and write data from a sqlite3 database.

Python Class

import sqlite3

class database_extensions():
    def __init__(self, db):
        self.databaseFileName = db

    def fetchAll(self, sql):
        """Fetch all of the records from the database"""
        conn = sqlite3.connect(self.databaseFileName)
        cursor = conn.cursor()
        cursor.execute(sql)
        records = cursor.fetchall()  
        conn.close() 
        return records   

    def fetchSingleValue(self, sql):
        """Fetch one single value from the database"""
        records = self.fetchAll(sql)  

        if not records and len(records) == 0: 
            raise Exception("No record found")

        if len(records) > 1: 
            raise Exception("More than one record found")

        return records[0][0]   

    def execute(self, sql):
        """Execute an sql command that will not return any records"""
        conn = sqlite3.connect(self.databaseFileName)
        cursor = conn.cursor()
        cursor.execute(sql)
        conn.commit()
        conn.close()

Unit Tests

The class database_extensions_tests contains a set of unit tests that will test the database_extensions functionality. The unit tests create a temporary database and populate the database with a table called settings.

import unittest
import tempfile
import sqlite3
import os
from database_extensions import database_extensions

class database_extensions_tests(unittest.TestCase):
    def setUp(self):
        """Set up for unit tests, function is executed before tests begin"""

        # Create a temporary file to use as a test database
        self.db_fd, self.db_path = tempfile.mkstemp()

        # Initialize the database with the test schema and data
        self.init_db()

    def tearDown(self):
        """Tear down for unit tests, function is executed after tests have been executed"""

        # Close and remove the temporary database
        os.close(self.db_fd)
        os.unlink(self.db_path)

    def init_db(self):
        """Initialize the database with the test schema and data"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('CREATE TABLE Settings (Sname TEXT, Svalue TEXT)')
        cursor.execute('INSERT INTO Settings (Sname, Svalue) VALUES (?, ?)', ('Setting1', 'Value1'))
        cursor.execute('INSERT INTO Settings (Sname, Svalue) VALUES (?, ?)', ('Setting2', 'Value2'))
        conn.commit()
        conn.close()

    def test_fetchAll(self):
        """Test confirms that FetchAll will return a list of all the system settings"""

        # Arrange 
        expected_result = [
            ('Setting1', 'Value1'),
            ('Setting2', 'Value2')
        ]
        systemUnderTest = database_extensions(self.db_path)

        # Act
        actual_result = systemUnderTest.fetchAll('SELECT Sname, Svalue FROM `Settings` ORDER BY Sname ASC')

        # Assert      
        self.assertEqual(actual_result, expected_result) 

    def test_fetchSingleValue(self):
        """Test confirms that fetchSingleValue will return a single system setting value"""

        # Arrange 
        expected_result = 'Value1'
        systemUnderTest = database_extensions(self.db_path)

        # Act
        actual_result = systemUnderTest.fetchSingleValue("SELECT Svalue FROM `Settings` where Sname='Setting1'")

        # Assert      
        self.assertEqual(actual_result, expected_result) 

    def test_fetchSingleValue_no_record_found(self):
        """Test confirms that fetchSingleValue raise exception when the value is not found"""

        # Arrange 
        systemUnderTest = database_extensions(self.db_path)

        # Act and Assert
        with self.assertRaises(Exception) as context:
            systemUnderTest.fetchSingleValue("SELECT Svalue FROM `Settings` where Sname='bad'")        
        self.assertEqual(str(context.exception), "No record found") # Assert that the exception message is "No record found"

    def test_fetchSingleValue_too_many_records_found(self):
        """Test confirms that fetchSingleValue raise exception when too many records are found"""

        # Arrange 
        systemUnderTest = database_extensions(self.db_path)

        # Act and Assert
        with self.assertRaises(Exception) as context:
            systemUnderTest.fetchSingleValue("SELECT Svalue FROM `Settings`")        
        self.assertEqual(str(context.exception), "More than one record found") # Assert that the exception message is "More than one record found"

    def test_execute(self):
        """Text confirms that execute can insert a record"""

        # Arrange 
        sname = 'Setting5'
        svalue = 'Value5'
        expected_result = [
            ('Setting1', 'Value1'),
            ('Setting2', 'Value2'),
            (sname, svalue)
        ]
        systemUnderTest = database_extensions(self.db_path)

        # Act
        systemUnderTest.execute(f"INSERT INTO `Settings` (Sname, Svalue) VALUES ('{sname}', '{svalue}')")

        # Assert
        actual_result = systemUnderTest.fetchAll('SELECT Sname, Svalue FROM `Settings` ORDER BY Sname ASC')
        self.assertEqual(actual_result, expected_result) 

if __name__ == '__main__':
    unittest.main()

How To Create a Python Requirements File

Creating and using a requirements.txt file in Python is a standard practice for managing project dependencies. This file lists all the Python packages your project depends on, along with their versions. Here’s a step-by-step guide on how to create and use it:

Creating a requirements.txt File

Install the Necessary Packages

First, install the packages your project needs. You can do this using pip:

pip install package_name

Freeze the Current Environment

Once you’ve installed all the necessary packages, you can generate a requirements.txt file using the pip freeze command. This command outputs a list of installed packages in the current environment and their versions.

pip freeze > requirements.txt

Manual Editing (Optional)

You can manually edit the requirements.txt file to add or remove packages, or to specify version ranges. The file format is simple: each line contains a package name and an optional version specifier.

package_name==1.0.0
another_package>=2.0,<3.0

Using a requirements.txt File

Create a Virtual Environment

It’s a good practice to use a virtual environment to manage your project’s dependencies. Create a virtual environment using:

python -m venv myenv

Activate the Virtual Environment

On Windows:

myenv\Scripts\activate

On macOS and Linux:

source myenv/bin/activate

Install Dependencies from requirements.txt

With the virtual environment activated, you can install all the packages listed in requirements.txt using:

pip install -r requirements.txt

Best Practices

Pin Versions: Pin the versions of your dependencies to ensure that your project works as expected regardless of updates to those packages. Use == to pin exact versions.

numpy==1.19.5
pandas==1.2.4

Update Regularly: Regularly update your requirements.txt file to keep track of changes in your dependencies.

pip freeze > requirements.txt

Review Dependencies: Periodically review your requirements.txt file to remove unnecessary dependencies that might have been included during development but are no longer needed.

Example Workflow

Install Dependencies

pip install numpy pandas

Freeze Dependencies

pip freeze > requirements.txt

Check the Generated requirements.txt

numpy==1.21.0
pandas==1.3.0

Create and Activate a Virtual Environment

python -m venv myenv
source myenv/bin/activate  # macOS/Linux
myenv\Scripts\activate     # Windows

Install Dependencies in the New Environment

pip install -r requirements.txt

By following these steps, you can ensure that your project’s environment is reproducible, making it easier to collaborate with others and deploy your application.

How To Secure A Flask Python Application

Introduction

Securing a Python Flask application with JWT authentication is crucial for safeguarding sensitive data and ensuring user privacy and security. JWT (JSON Web Tokens) authentication offers a lightweight, stateless mechanism for verifying the identity of clients accessing the application. By implementing JWT authentication, developers can prevent unauthorized access to resources and endpoints within the Flask application.

JWT authentication adds an extra layer of security by generating tokens that contain encoded user information, which can be securely transmitted between the client and the server. These tokens are signed with a secret key, making them tamper-proof and resistant to unauthorized alterations. With JWT, developers can enforce access controls, validate the integrity of incoming requests, and manage user sessions efficiently without relying on server-side storage.

Furthermore, JWT authentication enhances scalability and performance by eliminating the need for server-side session management, reducing the burden on server resources. This approach also promotes interoperability, allowing different services and platforms to authenticate users seamlessly using standardized JWT tokens. Overall, securing a Python Flask application with JWT authentication helps mitigate security risks, enhances user experience, and ensures compliance with modern security standards.

In this article, we are going to secure a simple Python Flask application backend and build a simple HTML frontend that will make requests to the backend.

The Backend

The backend Python Flask application hosts three endpoints.

  • GET /api/unsecure

    The Unsecured endpoint returns data without authorization.

  • GET /api/secure

    The secure endpoint requires authorization

  • POST /api/login

    The login endpoint validates a user’s credentials and returns an access token.

Creating The Backend Application

  • Create a text file called backend.py and enter the code from backend.py.
  • From the terminal, install the required libraries
pip install flask
pip install flask_restx
pip install flask_cors
pip install flask_jwt_extended

Testing The Backend Application

  • From the terminal execute the Python application python backend.py

  • The console output from the Python application will display the server address. In this example, the backend is hosted at http://127.0.0.1:5500
    start-backend

  • Open a web browser and navigate to http://127.0.0.1:5500

  • The web browser will show the Swagger page for the backend. Click api to display the endpoints.
    1

  • Test the unsecure endpoint

    • Click the GET next to /api/unsecure
      2
    • Click Try it out button
      3
    • Click Execute button
      4
    • The Response Body will display the JSON response from the GET request.
      5
  • Test the secure endpoint without authorization

    • Click the GET next to /api/secure
      6
    • Click Try it out button and then click the Execute button. The Response Body will display the message “Missing Authorization Header”.
      7
  • Generate Authorization Token

    • Click POST next to /api/login
      8
    • Click the Try it out button
      9
    • Enter the username kyle, password p1 and click the Execute button.
      10
    • Copy the access token from the Response Body.
      11
  • Test the secure endpoint with authentication

    • Click the padlock icon on the right hand side of GET /api/secure
      12
    • Enter Bearer into the value text box followed by the access token, then click the Authorize button.
      13
    • Click the Close button to close the authorization window.
      14
    • Click the Execute button, this time the Response Body should show the JSON data from the request.
      15

Endpoint Authentication

The main difference between the definitions for the secure and unsecure endpoints are the function decorators.

Unsecure endpoint

@ns.route('/unsecure')
class UnsecureData(Resource):
    def get(self):

Secure endpoint, note the function decorators @jwt_required() and @ns.doc(security='jsonWebToken'). The decorators enforce that the request header must contain the access token.

@ns.route('/secure')
class SecureData(Resource):
    @jwt_required()
    @ns.doc(security='jsonWebToken')
    def get(self):

The Frontend

Creating The Frontend Application

  • Create a text file called frontend.html and enter the code from frontend.html.

Testing The Frontend Application

  • Using a web browser open the file frontend.html
    16

  • Enter the username kyle, password p1 and then click the Log In button.
    17

  • Click the OK button to confirm the successful login.
    18

  • Click the Get Secure Data button, the JSON response data will be displayed below the button.
    19

Summary

In this solution, no user credentials are stored in the front end. The backend validates the user’s name and password and returns a temporary access token, the access token is used to authorize requests to the backend endpoints.

How To Read Settings From .env File Using Python Dotenv Module

The Dotenv Python module provides an easy way to read settings from a .env file.

  • Install the Python module
pip install python-dotenv
  • Create a text file called .env. In this example, the setting file will contain one setting called “myName” with the value “andrew”
myName=andrew
  • Create a Python file. The important parts of the script are
  • Import the module using from dotenv import load_dotenv.
  • Load the settings from the .env file using load_dotenv()
  • Get the value for the setting called myName using os.getenv("myName")
import os
from dotenv import load_dotenv

# Load the settings from the .env file
load_dotenv()

# Get the value for the setting myName
myName = os.getenv("myName")
print("MyName=", myName)