Introduction to ORM with Flask-SQLAlchemy

Written by Ashwin Joy in Python,Web Development

While Flask provides the essentials to get a web application up and running, it doesn’t force anything upon the developer. This means that many features aren’t included in the core framework. Instead, these features are available as extensions.

What are Flask Extensions?

Flask extensions are packages that add functionality to Flask. They integrate seamlessly with Flask, so they feel just like an integral part of the framework. They can add features, simplify complex operations, or even provide patterns and best practices.

Some Popular Flask Extensions:

  1. Flask-RESTful: Simplifies the creation of REST APIs.
  2. Flask-SQLAlchemy: Adds ORM capabilities, allowing for easier database interactions.
  3. Flask-WTF: Simplifies forms handling, including validation and rendering.

In this tutorial, we’ll focus on Flask-SQLAlchemy, an extension that provides ORM capabilities to Flask applications. Before we dive deep into Flask-SQLAlchemy, let’s first look at what ORM is.

What is ORM?

Object-Relational Mapping (ORM) is a programming technique that allows you to interact with your database, like you would with SQL. In other words, it maps your database tables to objects in code, allowing for more intuitive database operations.

Why use ORM?

  1. Abstraction: You don’t have to write raw SQL queries.
  2. Maintainability: Easier to update, maintain, and reuse code.
  3. Security: Helps prevent SQL injection attacks.
  4. Database Agnostic: Your code can work with different types of databases with minimal changes.

Getting Started with Flask-SQLAlchemy

Installation:

Install Flask and Flask-SQLAlchemy:

pip install flask flask-sqlalchemy

Initialize Flask and Flask-SQLAlchemy:

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'your_database_URI_here'
db = SQLAlchemy(app)

Replace 'your_database_URI_here' with the appropriate URI for your database (e.g., 'sqlite:///mydatabase.db' for SQLite).

Defining Models with Flask-SQLAlchemy

Models in Flask-SQLAlchemy represent tables in the database:

class YourModelName(db.Model):
    column_name = db.Column(Type, constraints)
    # ... other columns ...

In the above code,

  • YourModelName: Replace with the name of your model (e.g., Task).
  • column_name: Replace with the name of your column (e.g., id).
  • Type: The type of the column (e.g., db.Integer, db.String(80)).
  • constraints: Optional constraints or directives (e.g., primary_key=True).

Example:

class Task(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(80), nullable=False)
    description = db.Column(db.String(200))

To create the table, you need to run the following code:

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

Make sure you add with app.app_context()before your database operations in Flask. You can add this line before each of the CRUD operations as well.

CRUD Operations with Flask-SQLAlchemy

Create:

task = Task(title="Learn Flask", description="Study Flask-SQLAlchemy")
db.session.add(task)
db.session.commit()

Read:

All tasks:

tasks = Task.query.all()

Specific task:

task = db.session.get(Task, task_id)

Update:

task = db.session.get(Task, task_id)
task.title = "Learn Flask Extensions"
db.session.commit()

Delete:

task = db.session.get(Task, task_id)
db.session.delete(task)
db.session.commit()

Running the Application:

if __name__ == '__main__':
    app.run(debug=True)

Here’s the complete code for a simple Python script that demonstrates CRUD operations using Flask-SQLAlchemy without any user interface or interaction.

from flask_sqlalchemy import SQLAlchemy
from flask import Flask

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///mydatabase.db'
db = SQLAlchemy(app)

class Task(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(80), nullable=False)
    description = db.Column(db.String(200))

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

def add_task(title, description):
    with app.app_context():
        task = Task(title=title, description=description)
        db.session.add(task)
        db.session.commit()

def get_all_tasks():
    with app.app_context():
        return Task.query.all()

def update_task(task_id, new_title, new_description):
    with app.app_context():
        task = db.session.get(Task, task_id)
        if task:
            task.title = new_title
            task.description = new_description
            db.session.commit()

def delete_task(task_id):
    with app.app_context():
        task = db.session.get(Task, task_id)
        if task:
            db.session.delete(task)
            db.session.commit()

if __name__ == '__main__':
    create_tables()
    add_task("Learn Flask", "Study Flask-SQLAlchemy")
    add_task("Learn REST", "Study REST APIs")
    tasks = get_all_tasks()
    for task in tasks:
        print(task.id, task.title, task.description)
    update_task(2, "Learn Flask Extensions", "Study Flask-SQLAlchemy in depth")
    delete_task(1)

This is a good foundational example for someone learning Flask-SQLAlchemy. It showcases the basic operations without the complexities of a web interface or API endpoints.

CRUD Operations with Flask-SQLAlchemy and REST APIs

Let’s integrate Flask routes to provide RESTful API endpoints for the CRUD operations on the Task model.

Here’s the updated code:

from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///mydatabase.db'
db = SQLAlchemy(app)

class Task(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(80), nullable=False)
    description = db.Column(db.String(200))

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

@app.route('/tasks', methods=['POST'])
def create_task():
    with app.app_context():
        data = request.get_json()
        task = Task(title=data['title'], description=data.get('description', ''))
        db.session.add(task)
        db.session.commit()
        return jsonify({"message": "Task created!", "task_id": task.id}), 201

@app.route('/tasks', methods=['GET'])
def get_tasks():
    with app.app_context():
        tasks = Task.query.all()
        return jsonify([{'id': task.id, 'title': task.title, 'description': task.description} for task in tasks])

@app.route('/tasks/<int:task_id>', methods=['GET'])
def get_single_task(task_id):
    with app.app_context():
        task = db.session.get(Task, task_id)
        if not task:
            return jsonify({"message": "Task not found!"}), 404
        return jsonify({'id': task.id, 'title': task.title, 'description': task.description})

@app.route('/tasks/<int:task_id>', methods=['PUT'])
def update_task(task_id):
    with app.app_context():
        task = db.session.get(Task, task_id)
        if not task:
            return jsonify({"message": "Task not found!"}), 404
        data = request.get_json()
        task.title = data['title']
        task.description = data.get('description', '')
        db.session.commit()
        return jsonify({"message": "Task updated!"})

@app.route('/tasks/<int:task_id>', methods=['DELETE'])
def delete_task(task_id):
    with app.app_context():
        task = db.session.get(Task, task_id)
        if not task:
            return jsonify({"message": "Task not found!"}), 404
        db.session.delete(task)
        db.session.commit()
        return jsonify({"message": "Task deleted!"})

if __name__ == '__main__':
    create_tables()
    app.run(debug=True)

This code provides RESTful API endpoints for:

  • Creating a new task (POST /tasks)
  • Retrieving all tasks (GET /tasks)
  • Retrieving a single task by its ID (GET /tasks/<task_id>)
  • Updating a task by its ID (PUT /tasks/<task_id>)
  • Deleting a task by its ID (DELETE /tasks/<task_id>)

You can use tools like curl or Postman to interact with these endpoints.

Here are the curl commands to interact with the provided Flask application for CRUD operations:

  1. Create a New Task (POST /tasks)
curl -X POST -H "Content-Type: application/json" -d '{"title":"Learn Flask", "description":"Study Flask-SQLAlchemy"}' http://localhost:5000/tasks
  1. Retrieve All Tasks (GET /tasks)
curl http://localhost:5000/tasks
  1. Retrieve a Single Task by ID (GET /tasks/{task_id})
    Replace {task_id} with the actual ID of the task you want to retrieve.
curl http://localhost:5000/tasks/{task_id}
  1. Update a Task by ID (PUT /tasks/{task_id})
    Replace {task_id} with the actual ID of the task you want to update.
curl -X PUT -H "Content-Type: application/json" -d '{"title":"Updated Title", "description":"Updated Description"}' http://localhost:5000/tasks/{task_id}
  1. Delete a Task by ID (DELETE /tasks/{task_id})
    Replace {task_id} with the actual ID of the task you want to delete.
curl -X DELETE http://localhost:5000/tasks/{task_id}

Make sure your Flask application is running when you execute these commands. Adjust the localhost:5000 part if your application is running on a different host or port.

Final Thoughts

There you have it, Pythonistas! We’ve journeyed through the realms of Flask, dived deep into ORM with Flask-SQLAlchemy, and surfaced with RESTful APIs in our toolkit. As always, the world of Python offers endless possibilities, and this is just a stepping stone.

Keep experimenting, keep coding, and remember: every line of code is a step closer to mastery. Until next time, happy coding!

Ashwin Joy

I'm the face behind Pythonista Planet. I learned my first programming language back in 2015. Ever since then, I've been learning programming and immersing myself in technology. On this site, I share everything that I've learned about computer programming.

Leave a Reply

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

Recent Posts