Be Sure to Complete the Series

In this Flask web development series, we'll dive deeper into Flask and cover topics that build on the basics introduced in the first article. We'll explore more advanced features of Flask, including database integration, user authentication, and RESTful APIs. We'll also show you how to use Docker to containerize your Flask application, simplify deployment, and set up CI/CD with GitHub Actions to automate testing and deployment. By following this series, you'll gain a comprehensive understanding of Flask and its ecosystem, and easily build robust, scalable web applications. So follow the entire series to get the most out of your Flask learning journey.

Learning Outcomes

The learning outcome of the article is to demonstrate how to create a Flask application with a SQL or NoSQL database connection. By following the steps outlined in the article, readers will be able to build a web application that can retrieve and store data from a database. The benefits of having a database connection in an app are numerous. It allows for efficient storage and retrieval of data, enables data analysis and reporting, and facilitates scalability as the application grows in size and complexity. Additionally, a database connection can help to ensure data integrity and security by providing features such as user authentication and access control. Overall, learning how to create a Flask app with a database connection is a valuable skill for anyone interested in building robust and scalable web applications.

If you feel stuck please check the github repo to get some help.

Link: https://github.com/ritwikmath/Mastering-Flask/tree/article-two

SQL and NoSQL: The Dynamic Duo of Database Management

SQL and NoSQL are the two most popular types of databases used by organizations around the world. SQL, or Structured Query Language, is a relational database management system that uses tables to store data and uses SQL queries to manipulate and retrieve data. SQL is used by a variety of organizations, from small startups to large enterprises, and is especially popular for transactional and analytical workloads. On the other hand, NoSQL, or non-relational databases, use a variety of data models, including document-oriented, key-value, and graph databases, and are popular for their flexibility and scalability. NoSQL databases are often used by web applications and other systems that require a flexible and scalable data storage solution. Overall, SQL and NoSQL are the most widely used database systems due to their robustness, scalability, and flexibility, making them ideal for a wide range of applications and industries.

SQL Integration

This article specifically discusses the use of MySQL as the SQL database management system. MySQL is a popular relational database management system (RDBMS) that is often used in conjunction with SQL (Structured Query Language).

Do You Have These?

Do not use any special character '@' as user password. This can conflict with connection string's '@' sign. Instead of using root, create a seperate user and connect to it.

Problem Statement

To create a web-based application that allows users to perform CRUD (Create, Read, Update, Delete) operations on a 'developers' table. The 'developers' table could store information such as the developer's first name, last name, and area of expertise.

Configuration

Connect to MySQL localhost using Workbench. Create a database named 'flask'.

create database flask;

Add SQLAlchemy, cryptography, and pymysql to requirements.txt file. Install new packages in the flask virtual environment (follow the previous article).

pip install -r requirements.txt # install dependencies

Create a new file database.py. Create a class named Database.

import sqlalchemy

class Database:
    __engine = None
    __connection = None
    __user = 'ritwik'
    __pass = 'ronnie123'
    __host = 'localhost'
    __port = 3306
    __db = 'flask'

    def __new__(cls):
        if not hasattr(cls, 'instance'):
            cls.instance = super(Database, cls).__new__(cls)
        return cls.instance

    def connect(self):
        Database.__engine = sqlalchemy.create_engine(f"mysql+pymysql://{Database.__user}:{Database.__pass}@{Database.__host}/{Database.__db}?charset=utf8mb4&autocommit=true")
        Database.__connection = Database.__engine.connect()

    @property
    def connection(self):
        if not Database.__connection:
            raise RuntimeError('No database connection available.')
        return Database.__connection

    @property
    def engine(self):
        if not Database.__engine:
            raise RuntimeError('No database engine initialized.')
        return Database.__engine

Database class follows the singleton design pattern. This means no matter how many times, the class is initialized, all the time it will return one instance only. Without it, when a database sends a new instance with a null connection each time, it can lead to potential issues. This is because, for each instance, a new connection is made to the database, which can cause performance degradation and resource exhaustion over time. In contrast, the Singleton pattern can help alleviate these issues by ensuring that only one instance of the database connection is created and shared among all the instances of the application or program. This approach can help improve performance and reduce resource usage by avoiding the overhead of creating a new connection for each instance.

The connect method is used to establish a connection to the database using the SQLAlchemy create_engine function. This function takes a connection string as an argument and returns an Engine object that can be used to interact with the database.

Each attribute is a private and class attribute that prevents updating the values by mistake. The @property decorator is used to define the connection and engine methods as properties of the Database class. These properties are read-only and allow the database connection and engine to be accessed like attributes of the class.

The connection property returns the database connection object, and raises a RuntimeError if the connection has not been established.

The engine method in the Database class returns the database engine that is used to connect to the database. It first checks if the __engine attribute in the class has been initialized or not, and raises a RuntimeError if it has not been initialized. If it has been initialized, it returns the engine.

from flask import Flask, request
from database import Database
from sqlalchemy import Table, Column, Integer, String, MetaData, insert
import json

app = Flask(__name__)

meta = MetaData()

developers = Table(
   'developers', meta, 
   Column('id', Integer, primary_key = True), 
   Column('first_name', String(255)), 
   Column('last_name', String(255)), 
   Column('expert', String(255)), 
)

@app.post('/')
def create():
    return f''

@app.get('/')
def fetch():
    return f''

@app.patch('/<int:id>')
def update(id):
    return f''

@app.delete('/<int:id>')
def delete(id):
    return f''

if __name__ == '__main__':
    db = Database()
    db.connect()
    meta.create_all(db.engine)
    app.run(debug=True)

The Table, Column, and MetaData classes are imported from the sqlalchemy module, which is used to define the structure of the database table. A new MetaData instance is created using the MetaData class and stored in the meta variable. A new 'developers' table is defined using the Table class and the meta metadata object. The 'developers' table has three columns: id, first_name, last_name, and expert.

The application has four routes defined using the @app decorator:

• The '/' route with a POST method will call the create() function to create a new developer entry in the database.

• The '/' route with a GET method will call the fetch() function to retrieve all the developer entries from the database.

• The '/int:id' route with a PATCH method will call the update() function to update the developer entry with the specified ID.

• The '/int:id' route with a DELETE method will call the delete() function to delete the developer entry with the specified ID.

When the application runs, it creates an instance of the Database class and connects to the database. It then uses the MetaData instance to create the 'developers' table in the database. Finally, it starts the Flask application in debug mode.

CRUD using SQLAlchemy

@app.post('/')
def create():
    try:
        db = Database()
        ins = insert(developers).values(first_name = request.json.get('first_name'), 
                                last_name = request.json.get('last_name'),
                                expert = request.json.get('expert'))
        result = db.connection.execute(ins)
        return {'status': True, 'data': result.rowcount}
    except Exception as ex:
        return {'status': False, 'error': ex.__str__()}

The create() function is decorated with @app.post('/'), which registers it as a handler for HTTP POST requests to the root URL. The function first creates a new Database instance and connects to the database. It then creates an INSERT statement using the insert method from SQLAlchemy, passing the table object developers and the values to be inserted from the JSON payload of the request. The statement is executed using the database connection and the ResultProxy object is returned in the result variable. The function returns a JSON response containing a boolean status field indicating whether the operation was successful or not, and the rowcount attribute of the result object as a string. If an exception occurs, the response contains the exception message in the data field.

@app.get('/')
def fetch():
    try:
        db = Database()
        ins = developers.select()
        result = db.connection.execute(ins)
        developers_list = []
        for row in result:
            developer_dict = {'id': row.id, 'first_name': row.first_name, 'last_name': row.last_name, 'expert': row.expert}
            developers_list.append(developer_dict)
        return {'status': True, 'data': developers_list}
    except Exception as ex:
        return {'status': False, 'error': ex.__str__()}

The fetch() function is decorated with @app.get('/'), which registers it as a handler for HTTP GET requests to the root URL. The function first creates a new Database instance and connects to the database. It then creates a SELECT statement using the select method from SQLAlchemy, passing the developers table object. The statement is executed using the database connection and the ResultProxy object is returned in the result variable. The function initializes an empty list developers_list to store the result. It then iterates over the rows in the result object, creates a dictionary for each row containing the required fields, and appends it to developers_list. Finally, the function returns a JSON response containing a boolean status field indicating whether the operation was successful or not, and the list of developers as the data field. If an exception occurs, the response contains the exception message in the data field.

@app.patch('/<int:id>')
def update(id):
    try:
        db = Database()
        ins = developers.update().where(developers.c.id == id).values(**request.json)
        db.connection.execute(ins)
        updated_row = db.connection.execute(developers.select().where(developers.c.id == id))
        developers_list = []
        for row in updated_row:
            developer_dict = {'id': row.id, 'first_name': row.first_name, 'last_name': row.last_name, 'expert': row.expert}
            developers_list.append(developer_dict)
        return {'status': True, 'data': developers_list[0]}
    except Exception as ex:
        return {'status': False, 'error': ex.__str__()}

This is a Flask route that handles the HTTP PATCH request at the URL '/int:id'. It expects an integer value to be passed in the URL which will be stored in the variable 'id'. The route updates a record in the 'developers' table based on the value of 'id'. The update query is executed using SQLAlchemy's 'update' method with the 'where' clause to match the record based on the value of 'id'. The 'values' method is used to update the columns with the values passed in the request body. After the update is done, the updated record is fetched using a SELECT query with the 'id' value to get the latest values. The updated record is then converted to a dictionary and returned as a response in JSON format. If any error occurs during the process, an error message is returned instead.

@app.delete('/<int:id>')
def delete(id):
    try:
        db = Database()
        ins = developers.delete().where(developers.c.id == id)
        db.connection.execute(ins)
        return {'status': True, 'data': {'deleted_id': id}}
    except Exception as ex:
        return {'status': False, 'error': ex.__str__()}

The data that should be returned in the response of a delete API request depends on the specific requirements of your application. Generally, a delete API request would return a success status message along with any relevant information about the deleted resource. For example, you could return a JSON object with a 'status' key that has a value of 'success', and a 'data' key that contains the ID of the deleted resource. The 'data' key could also include any other relevant information about the deleted resource, such as its name or other attributes. Alternatively, if there was an error during the deletion process, you could return an error message with details about the error in the 'data' key.

NoSQL Integration

In the article, MongoDB is used as a NoSQL database. MongoDB is a popular NoSQL (non-relational) document-oriented database that is designed for scalability and flexibility. It stores data in a flexible, JSON-like format called BSON (Binary JSON) that allows for easy data retrieval and manipulation. It is often used in web development and other applications that require a flexible, scalable database solution.

Do You Have These?

In MongoDB there is no need to create the database or table before performing some operations on those. MongoDB creates those on runtime.

Before starting with MongoDB, a new folder named database is created. The database.py file is renamed into mysql.py and moved into database folder.

Problem Statement

To store activity, request, and error logs in MongoDB could be to develop a scalable and reliable system that can handle a large volume of logs generated by a web application. The system should be able to store the logs efficiently, retrieve them quickly, and provide various analytics and reporting capabilities.

The schema-less nature of MongoDB makes it advantageous for storing distinct data sets for different logs, as it allows for flexible and dynamic data structures without the need for a pre-defined schema.

Configure

from pymongo import MongoClient

class Database:
    __client = None
    __connection = None
    __host = 'localhost'
    __port = 27017
    __db = 'flask'

    def __new__(cls):
        if not hasattr(cls, 'instance'):
            cls.instance = super(Database, cls).__new__(cls)
        return cls.instance

    def connect(self):
        Database.__client = MongoClient(f'mongodb://{self.__host}:{self.__port}')
        Database.__connection = Database.__client[Database.__db]

    @property
    def db(self):
        if Database.__connection is not None:
            return Database.__connection
        raise RuntimeError('No database connection available.')

    @property
    def client(self):
        if not Database.__client:
            raise RuntimeError('No database engine initialized.')
        return Database.__client

Use the same singleton design pattern approach for creating the MongoDB database connection class. The 'db' method is the same as the 'connection' method in 'mysql.py'. However, in this case, the condition has to be 'is not None' instead of 'not Database.__connection' because Database objects do not implement truth value testing or bool(). Additionally, there is no need to mention the user and password for the localhost connection.

CRUD using PyMongo

@app.before_request
def logRequest():
    Mongo().db.logs.insert_one({
        'type': 'request',
        'url': request.url,
        'http_method': request.method,
        'body': request.method in ['POST', 'PATCH'] and request.get_json() or None
    })

@app.before_request is decorator that executes a function before the API request reaches the end point. The logRequest function stores the request log in logs collection. The document contains the following fields:

• "type": set to "request", indicating that this is a request log.

• "url": the URL of the incoming request.

• "http_method": the HTTP method of the request, such as "GET", "POST", "PATCH", etc.

• "body": the body of the request, if the request method is "POST" or "PATCH" and the body is in JSON format. If the request method is not "POST" or "PATCH", the "body" field is set to None.

@app.post('/')
def create():
    try:
        db = MySql()
        ins = insert(developers).values(first_name = request.json.get('first_name'), 
                                last_name = request.json.get('last_name'),
                                expert = request.json.get('expert'))
        result = db.connection.execute(ins)
        Mongo().db.logs.insert_one({
            'type': 'activity',
            'url': request.url,
            'function': 'create developer'
        })
        return {'status': True, 'data': result.rowcount}
    except Exception as ex:
        Mongo().db.logs.insert_one({
            'type': 'error',
            'url': request.url,
            'message': ex.__str__()
        })
        return {'status': False, 'error': ex.__str__()}

For each request there are two more logs insert operations. The insert_one method in try block stores the activity log. One in the except block store error log. The data stored in three of the cases are completely different.

@app.get('/logs')
def fetchAlllogs():
    try:
        logs = list(Mongo().db.logs.find({}))
        return {'status': True, 'data': json.loads(json.dumps(logs, default=str))}
    except Exception as ex:
        return {'status': False, 'error': ex.__str__()}

The function fetchAlllogs is executed when a GET request is received at the /logs endpoint. The function attempts to retrieve all logs from the MongoDB database by calling the find method on the logs collection. If the database query is successful, the logs are returned as a JSON response with a 'status' key set to True and the 'data' key set to a JSON-serialized version of the logs retrieved from the database. If the database query fails, an error message is returned as a JSON response with a 'status' key set to False and the 'error' key set to the exception message.

@app.put('/logs/<string:doc_id>')
def softDeletLog(doc_id):
    try:
        updated_doc = Mongo().db.logs.find_one_and_update({'_id': ObjectId(doc_id)},
                                   {
                                        '$set':  {'deleted': True}
                                   }, upsert=True, new=True)
        return {'status': True, 'data': json.loads(json.dumps(updated_doc, default=str))}
    except Exception as ex:
        return {'status': False, 'error': ex.__str__()}

The function softDeletLog is executed when a PUT request is received at the /logs/<doc_id> endpoint. The function attempts to update a log document in the MongoDB database by calling the find_one_and_update method on the logs collection. It uses the id field with the ObjectId constructor to match the specified docid. The $set operator is used to set the deleted field of the matching document to True. The upsert=True argument specifies that a new document should be created if a document with the specified _id is not found. The new=True argument specifies that the updated document should be returned instead of the original document. If the database update is successful, the updated document is returned as a JSON response with a 'status' key set to True and the 'data' key set to a JSON-serialized version of the updated document retrieved from the database. If the database update fails, an error message is returned as a JSON response with a 'status' key set to False and the 'error' key set to the exception message.

@app.delete('/logs/<string:doc_id>')
def deletLog(doc_id):
    try:
        Mongo().db.logs.delete_one({'_id': ObjectId(doc_id)})
        return {'status': True, 'data': {'deleted_id': doc_id}}
    except Exception as ex:
        return {'status': False, 'error': ex.__str__()}

The function deletLog is executed when a DELETE request is received at the /logs/<doc_id> endpoint. The function attempts to delete a log document from the MongoDB database by calling the delete_one method on the logs collection. It uses the *id field with the ObjectId constructor to match the specified doc_*id. If the database deletion is successful, a JSON response is returned with a 'status' key set to True and the 'data' key set to a dictionary containing the ID of the deleted document. If the database deletion fails, an error message is returned as a JSON response with a 'status' key set to False and the 'error' key set to the exception message.

Conclusion

In conclusion, this article provides a comprehensive overview of using SQLAlchemy to establish a connection with MySQL and PyMongo to connect with MongoDB. The article also highlights the importance of logging in to web applications and demonstrates how to implement logging using Python's Flask framework and MongoDB as a data store. Furthermore, the article sheds light on the key differences between a structured database like MySQL and a schemaless database like MongoDB. The concept of schemaless databases is explained in detail, and it is highlighted how they differ from structured databases in terms of data modeling and flexibility. Overall, this article is a valuable resource for developers who want to learn about database connectivity in Python, logging, and the differences between structured and schemaless databases.