Python Flask SQLAlchemy One-to-Many Example (2024)

In this article, we'll demonstrate how to create a Python One To Many relationship example by using Flask and SQLAlchemy which will create below rest endpoints for POST, GET, PUT, DELETE.

Python POST, GET, PUT, DELETE Rest Api's

1. POST - Create Employee/Address Record
2. GET - List all employees/address
3. GET - Get employees/address by it's id
4. PUT - Update/Edit selected employee details
5. DELETE - Remove selected employee record

The one-to-many relationship, which enables one entity to have multiple related entities, will be implemented using SQLAlchemy.

The example will show how each employee may have different addresses, such as their home, office, or permanent. The Flask environment will be set up, the database models will be defined, the relevant routes will be created, and then the CRUD operations will be carried out.

One-to-Many

One End - The relationship property is specified on the Model at one end.

addresses = db.relationship("AddressModel", backref="employees")

Many End- The ForeignKey is declared at the many end.

emp_id = db.Column(db.Integer, db.ForeignKey("employees.id"),  nullable=False)

Q: What is Flask Python?
Ans:

A Python module that makes it simple to create web applications is called Flask. Its core is compact and simple to extend; it's a microframework that has lots of features, such as url routing and a template engine.

What is virtual environment in Python?

A virtual environment virtualenv is used to create independent, separate virtual environments with their own dependencies for multiple project. It generates a new folder for the new virtual environment which has its own libraries folder, pip, and Python interpreter for the Python version used to activate the environment.

Prerequisites

1. Python 3.x
2. Flask
3. SQLAlchemy

Follow the steps listed below to begin the project:

1. Follow the steps to Install Python 3 and editor.
2. Create a new virtual environment
3. Select Python Interpreter
4. Use new virtual environment for the python project
5. Activate your new created virtualenv
6. Install Flask if not available

Q: What is Flask-SQLAlchemy?
Ans:

SQLAlchemy is an ORM library helps convert Python classes/models into database tables and objects into rows and fields. A Flask addon called Flask-SQLAlchemy allows in integrating SQLAlchemy with Flask applications.

Install SQLAlchemy

Project Structure

According to the project's use case or requirements, we can create files and folders.

Python Flask SQLAlchemy Model

1. Create db.py file and initialize the SQLAlchemy as given below:

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

2. Create model folder and __init__.py file with below code.

from models.employee import EmployeeModel
from models.address import AddressModel

3. Now create employee.py file under model folder. Initialize employee fields with its data type and it's constraints.

from db import db

class EmployeeModel(db.Model):
    __tablename__ = "employees"

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), unique=True, nullable=False)
    role = db.Column(db.String(80), unique=True, nullable=False)

    addresses = db.relationship("AddressModel", backref="employees")

4. Create address.py file under model folder. Initialize address fields with its data type and it's constraints.

from db import db

class AddressModel(db.Model):
    __tablename__ = "address"

    id = db.Column(db.Integer, primary_key=True)
    addr = db.Column(db.String(80), unique=True, nullable=False)
    type = db.Column(db.String(80), unique=True, nullable=False)

    emp_id = db.Column(db.Integer, db.ForeignKey("employees.id"),  nullable=False)

Python Flask - Create Marshmallow Schemas

Python Marshmallow library is used for object/data validation and serialization/deserialization. Marshmallow can also be used with Flask-SQLAlchemy.

• Create a new schemas.py file under project to define the model schema for validating the input data.

from marshmallow import Schema, fields

class AddressSchema(Schema):
    id = fields.Str(dump_only=True)
    addr = fields.Str(required=True)
    type = fields.Str(required=True)
    emp_id = fields.Str(dump_only=True)


class AddressUpdateSchema(Schema):
    addr = fields.Str()
    type = fields.Str()


class EmployeeSchema(Schema):
    id = fields.Str(dump_only=True)
    name = fields.Str(required=True)
    role = fields.Str(required=True)
    addresses = AddressSchema

class EmployeeUpdateSchema(Schema):
    name = fields.Str()
    role = fields.Str()

Python Flask Resources using Blueprint

1. Install the Flask Smorest using pip install flask-smorest command in order to utilize blueprint functionality.
2. Create the blueprint for the employee : resources/employee.py
3. Create the blueprint for the address : resources/address.py
4. Import the blueprint from flask-smorest.
5. Add MethodViews for GET, POST, PUT and DELETE endpoints.

employee.py

from flask.views import MethodView
from flask_smorest import Blueprint, abort
from sqlalchemy.exc import SQLAlchemyError
from db import db
from models import EmployeeModel
from schemas import EmployeeSchema, EmployeeUpdateSchema

empblueprint = Blueprint("Employees", "employees", description="TechGeekNext - Employee CRUD operations")


@empblueprint.route("/employee/<string:emp_id>")
class Employee(MethodView):
    @empblueprint.response(200, EmployeeSchema)
    def get(self, emp_id):
        emp = EmployeeModel.query.get_or_404(emp_id)
        return emp

    def delete(self, emp_id):
        emp = EmployeeModel.query.get_or_404(emp_id)
        db.session.delete(emp)
        db.session.commit()
        return {"message": "Employee deleted."}

    @empblueprint.arguments(EmployeeUpdateSchema)
    @empblueprint.response(200, EmployeeSchema)
    def put(self, emp_data, emp_id):
        emp = EmployeeModel.query.get(emp_id)

        if emp:
            emp.name = emp_data["name"]
            emp.role = emp_data["role"]
        else:
            emp = EmployeeModel(id=emp_id, **emp_data)

        db.session.add(emp)
        db.session.commit()

        return


@empblueprint.route("/employees")
class GetEmployees(MethodView):
    @empblueprint.response(200, EmployeeSchema(many=True))
    def get(self):
        emp = EmployeeModel.query.all()
        return emp


@empblueprint.route("/employee")
class CreateEmp(MethodView):
    @empblueprint.arguments(EmployeeSchema)
    @empblueprint.response(201, EmployeeSchema)
    def post(self, emp_data):
        emp = EmployeeModel(**emp_data)
        print(emp)
        try:
            db.session.add(emp)
            db.session.commit()
        except SQLAlchemyError:
            abort(500, message="An error occurred while creating an employee.")

        return

address.py

from flask.views import MethodView
from flask_smorest import Blueprint, abort
from sqlalchemy.exc import SQLAlchemyError
from db import db
from models import AddressModel
from schemas import AddressSchema, AddressUpdateSchema

empblueprint = Blueprint("Address", "address", description="TechGeekNext - Address CRUD operations")


@empblueprint.route("/employee/<string:emp_id>/address")
class EmployeeAdd(MethodView):

    @empblueprint.arguments(AddressSchema)
    @empblueprint.response(201, AddressSchema)
    def post(self, address_data, emp_id):
        addr = AddressModel(**address_data, emp_id=emp_id)
        try:
            db.session.add(addr)
            db.session.commit()
        except SQLAlchemyError:
            abort(500, message="An error occurred while creating an address.")

        return


@empblueprint.route("/address/<string:add_id>")
class Address(MethodView):
    @empblueprint.response(200, AddressSchema)
    def get(self, add_id):
        return AddressModel.query.get_or_404(add_id)


    def delete(self, add_id):
        emp = AddressModel.query.get_or_404(add_id)
        db.session.delete(emp)
        db.session.commit()
        return {"message": "Address deleted."}

    @empblueprint.arguments(AddressUpdateSchema)
    @empblueprint.response(200, AddressSchema)
    def put(self, address_data, add_id):
        addr = AddressModel.query.get(add_id)

        if addr:
            addr.name = address_data["address"]
            addr.type = address_data["type"]
        else:
            addr = AddressModel(id=add_id, **address_data)

        db.session.add(addr)
        db.session.commit()

        return


@empblueprint.route("/address")
class GetAllAddress(MethodView):
    @empblueprint.response(200, AddressSchema(many=True))
    def get(self):
        return AddressModel.query.all()

Flask application factory

1. We have been directly creating the app variable in the Flask app called app.py.
2. We create a create_app() method that returns an app using the app factory pattern. So that we can configure the app before receiving it back, we can give configuration values to the function in this manner.
3. Provide the Swagger documentation details.
4. Configure the database connection with the app.
5. Connect the database using the db.init_app() method.
6. Register the Employee Blueprint.
7. Register the Address Blueprint.

from flask import Flask
from flask_smorest import Api
from db import db
from resources.employee import empblueprint as EmployeeBlueprint
from resources.address import empblueprint as AddressBlueprint

def create_app():
    # creates an application instance app using Flask() class
    app = Flask(__name__)

    # swagger documentation details
    app.config["PROPAGATE_EXCEPTIONS"] = True
    app.config["API_TITLE"] = "Employee Crud Example"
    app.config["API_VERSION"] = "v1"
    app.config["OPENAPI_VERSION"] = "3.0.3"
    app.config["OPENAPI_URL_PREFIX"] = "/"
    app.config["OPENAPI_SWAGGER_UI_PATH"] = "/swagger-ui"
    app.config[
        "OPENAPI_SWAGGER_UI_URL"
    ] = "https://cdn.jsdelivr.net/npm/swagger-ui-dist/"

    # Database configuration - connection
    app.config["SQLALCHEMY_DATABASE_URI"] = "mysql+pymysql://root:root@localhost/employeetestdb"

    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False

    # Connect the database to the app instance
    db.init_app(app)

    api = Api(app)

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

    #register employee blueprint
    api.register_blueprint(EmployeeBlueprint)

    # register address blueprint
    api.register_blueprint(AddressBlueprint)

    return app

Python Flask CRUD Example - POST, PUT, GET, DELETE endpoints

1. Open the database and create the schema using below SQL command.

CREATE SCHEMA `employeetestdb` ;

2. Run the python application using flask run command.


3. Create New Employee

   Open Postman, use POST method with end point http://localhost:5000/employee and provide Employee details to create new employee record.

4. List Employees

   Use GET method with end point http://localhost:5000/employees to get all employees.

5. Create Address

   Use POST method with end point http://localhost:5000/employee/1/address, where 1 is the employee id to be linked with address.

6. Get Addresses

   Use GET method with end point http://localhost:5000/address to get all address.
7. We can see that the data in database tables updated using all above CRUD operations.

Download Source Code

The full source code for this article can be found on below.
Download it here - Python Flask SQLAlchemy One-to-Many Example