rakesh kumar

Posted on Aug 16

How to perform crud in flask

Crud operation Using postman

Crud operation without postman

Importing Required Libraries

Flask: The main framework used to build the web application.
jsonify: A Flask utility that serializes data to JSON format, making it easier to send JSON responses.
request: A Flask utility used to handle incoming HTTP requests, particularly useful for extracting data from POST requests.
SQLAlchemy: An ORM (Object-Relational Mapper) that allows you to interact with the database using Python objects instead of writing raw SQL.it creates table in database using frontend

Initializing the Flask Application

app = Flask(__name__)

This line initializes the Flask application. The name variable helps Flask determine the root path of the application.

Configuring the Database Connection

app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://postgres:root@localhost:5433/flask_database'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

SQLALCHEMY_DATABASE_URI: This configuration sets up the database connection. It uses the PostgreSQL adapter psycopg2 to connect to a PostgreSQL database.
postgresql:// specifies the database dialect.
postgres:root@localhost:5433 provides the username (postgres), password (root), host (localhost), and port (5433) where the database is running.
/flask_database specifies the name of the database to connect to.
SQLALCHEMY_TRACK_MODIFICATIONS: This disables the feature that tracks modifications of objects and emits signals, saving memory.

Initializing SQLAlchemy

db = SQLAlchemy(app)

This line initializes SQLAlchemy with the Flask application, allowing you to define models and interact with the database.

Defining the Database Model

class Task(db.Model):
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    title = db.Column(db.String(200), nullable=False)
    done = db.Column(db.Boolean, default=False)

Task: This is a model class that represents a table in the database.
id: An integer column that serves as the primary key, which uniquely identifies each record. It auto-increments with each new task.
title: A string column (with a maximum of 200 characters) that stores the title of the task. It's marked as nullable=False, meaning it cannot be empty.
done: A boolean column that indicates whether the task is completed. It defaults to False.
Creating the Database Tables

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

This code ensures that the database tables are created based on the defined models. create_all() will create the tables if they don’t already exist. app.app_context() is used to provide the necessary application context required by SQLAlchemy.

Defining Routes (Endpoints) Hello World Route

@app.route('/')
def hello_world():
    return jsonify({"key": "hello"})

OUTPUT

Purpose: This is a simple route that returns a JSON response {"key": "hello"} when the root URL (/) is accessed.
Get All Tasks

@app.route('/tasks')
def get_tasks():
    tasks = Task.query.all()
    task_list = [{'id': task.id, 'title': task.title, 'done': task.done} for task in tasks]
    return jsonify({"tasks": task_list})

Purpose: This route fetches all tasks from the Task table and returns them as a list of dictionaries in JSON format.
Task.query.all(): Retrieves all records from the Task table.
task_list: A list comprehension that converts each Task object into a dictionary containing its id, title, and done status.
Create a New Task

@app.route('/tasks', methods=['POST'])
def create_task():
    data = request.get_json()
    new_task = Task(title=data['title'], done=data['done'])
    db.session.add(new_task)
    db.session.commit()
    return jsonify({'message': 'Task created'})

Purpose: This route allows you to create a new task by sending a POST request with a JSON payload containing the title and done status of the task.
request.get_json(): Parses the incoming JSON data from the request.
db.session.add(new_task): Adds the new task to the current database session.
db.session.commit(): Commits the transaction to the database, saving the new task.
Response: Returns a message indicating that the task was successfully created.
Update an Existing Task

@app.route('/tasks/<int:id>', methods=['PUT'])
def update_task(id):
    task = Task.query.get(id)
    if task is None:
        return jsonify({'message': 'Task not found'}), 404

    data = request.get_json()
    task.title = data.get('title', task.title)
    task.done = data.get('done', task.done)
    db.session.commit()

    return jsonify({'message': 'Task updated'})

Purpose: This route allows you to update an existing task by its id. The task’s title and done status can be modified.
int:id: The route takes an integer id parameter to identify the specific task to update.
Task.query.get(id): Retrieves the task with the specified id from the database.
task.title = data.get('title', task.title): Updates the title if provided in the request; otherwise, it keeps the current title.
db.session.commit(): Commits the changes to the database.
Response: Returns a message indicating that the task was successfully updated.
Delete a Task

@app.route('/tasks/<int:id>', methods=['DELETE'])
def delete_task(id):
    task = Task.query.get(id)
    if task is None:
        return jsonify({'message': 'Task not found'}), 404

    db.session.delete(task)
    db.session.commit()

    return jsonify({'message': 'Task deleted'})

Purpose: This route allows you to delete an existing task by its id.
Task.query.get(id): Retrieves the task with the specified id from the database.
db.session.delete(task): Marks the task for deletion.
db.session.commit(): Commits the deletion to the database.
Response: Returns a message indicating that the task was successfully deleted.
Running the Application

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

This block ensures that the Flask application runs only if the script is executed directly. The debug=True argument enables debug mode, which provides useful error messages and auto-reloads the server when code changes are detected.
Summary:
Database Connection: The application connects to a PostgreSQL database using SQLAlchemy, which is configured via the SQLALCHEMY_DATABASE_URI.
CRUD Operations: The application allows you to perform CRUD (Create, Read, Update, Delete) operations on tasks via different HTTP methods (GET, POST, PUT, DELETE).
Flask Routes: Each route handles a specific task-related operation, interacting with the database through the SQLAlchemy ORM.

Full Code

app.py

from flask import Flask,jsonify,request
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)

app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://postgres:root@localhost:5433/flask_database'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

# Initialize the database
db = SQLAlchemy(app)

class Task(db.Model):
    # __table__ = 'tasks'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    title = db.Column(db.String(200), nullable=False)
    done = db.Column(db.Boolean, default=False)

with app.app_context():
    db.create_all()
@app.route('/')
def hello_world():  # put application's code here
    # return jsonify('hello')
    return jsonify({"key":"hello"})
    # return ('hello')

@app.route('/tasks')
def get_tasks():
    tasks = Task.query.all()
    task_list = [
        {'id': task.id, 'title': task.title, 'done': task.done} for task in tasks
    ]
    return jsonify({"tasks": task_list})
@app.route('/tasks', methods=['POST'])
def create_task():
    data = request.get_json()
    new_task = Task(title=data['title'], done=data['done'])
    db.session.add(new_task)
    db.session.commit()
    return jsonify({'message': 'Task created'})

@app.route('/tasks/<int:id>', methods=['PUT'])
def update_task(id):
    task = Task.query.get(id)
    if task is None:
        return jsonify({'message': 'Task not found'}), 404

    data = request.get_json()
    task.title = data.get('title', task.title)
    task.done = data.get('done', task.done)
    db.session.commit()

    return jsonify({'message': 'Task updated'})

@app.route('/tasks/<int:id>', methods=['DELETE'])
def delete_task(id):
    task = Task.query.get(id)
    if task is None:
        return jsonify({'message': 'Task not found'}), 404

    db.session.delete(task)
    db.session.commit()

    return jsonify({'message': 'Task deleted'})

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

Crud operation without postman

my folder structure

Step 2: Create the Flask App Factory (init.py)
The app factory pattern allows you to create multiple instances of your app, which is useful for testing. Here's how to define the app factory:

init.py:

from flask import Flask
from .config import Config
from .extensions import db
from .views import main


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

    # Initialize extensions
    db.init_app(app)

    # Register blueprints
    app.register_blueprint(main)

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

    return app

Step 3: Configuration File (config.py)
Store your configuration settings, such as the database URI, in a separate file:

config.py:

class Config:
    SQLALCHEMY_DATABASE_URI = 'postgresql://postgres:root@localhost:5433/flask_database'
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    SECRET_KEY = 'your_secret_key'

Step 4: Extensions Initialization (extensions.py)
Create a separate file to initialize your extensions, such as SQLAlchemy:

extensions.py:

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

Step 5: Define Models (models.py)
Define your database models in a separate file:

models.py:

from .extensions import db

class Task(db.Model):
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    title = db.Column(db.String(200), nullable=False)
    done = db.Column(db.Boolean, default=False)

Step 6: Define Views and Routes (views.py)
Create your view functions and routes in a separate file:

views.py:

from flask import Blueprint, jsonify, request, redirect, url_for, render_template
from .models import Task
from .extensions import db

# Create a Blueprint
main = Blueprint('main', __name__)

# Route to fetch all tasks
# Route to render the index page
@main.route('/')
def index():
    tasks = Task.query.all()
    return render_template('index.html', tasks=tasks)
@main.route('/tasks')
def get_tasks():
    tasks = Task.query.all()
    task_list = [{'id': task.id, 'title': task.title, 'done': task.done} for task in tasks]
    return jsonify({"tasks": task_list})

# Route to create a new task
@main.route('/tasks', methods=['POST'])
def create_task():
    title = request.form['title']
    new_task = Task(title=title, done=False)
    db.session.add(new_task)
    db.session.commit()
    return redirect(url_for('main.index'))
# @app.route('/tasks', methods=['POST'])
# def create_task():
#     data = request.get_json()
#     new_task = Task(title=data['title'], done=data['done'])
#     db.session.add(new_task)
#     db.session.commit()
#     return jsonify({'message': 'Task created'}


# Route to update an existing task
@main.route('/tasks/<int:id>', methods=['PUT'])
def update_task(id):
    task = Task.query.get(id)
    if task is None:
        return jsonify({'message': 'Task not found'}), 404

    data = request.get_json()
    task.title = data.get('title', task.title)
    task.done = data.get('done', task.done)
    db.session.commit()

    return jsonify({'message': 'Task updated'})

# Route to delete a task
@main.route('/tasks/<int:id>', methods=['DELETE'])
def delete_task(id):
    task = Task.query.get(id)
    if task is None:
        return jsonify({'message': 'Task not found'}), 404

    db.session.delete(task)
    db.session.commit()

    return jsonify({'message': 'Task deleted'})

Step 7: Entry Point (app.py)
Your app.py file should now simply call the app factory to create the Flask application:

app.py:

from . import create_app

app = create_app()

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

Step 8: Running the Application
To run your Flask application, use the following command in your project directory:

python -m app

Set Up the Templates
Create your HTML templates to allow users to input data and display tasks.

templates/layout.html: Base layout for your HTML files.

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Flask CRUD App</title>
    <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css">
</head>
<body>
    <div class="container mt-4">
        <h1>Flask CRUD App</h1>
        {% block content %}{% endblock %}
    </div>
</body>
</html>

templates/index.html: Home page with a form to input the task title.

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Task List</title>
</head>
<body>
    <h1>Task List</h1>
    <ul>
        {% for task in tasks %}
        <li>{{ task.title }} - {% if task.done %}Done{% else %}Not Done{% endif %}</li>
        {% endfor %}
    </ul>

    <h2>Create a New Task</h2>
    <form method="POST" action="{{ url_for('main.create_task') }}">
        <label for="title">Task Title:</label>
        <input type="text" name="title" id="title">
        <button type="submit">Add Task</button>
    </form>
</body>
</html>

output

Summary
This structure organizes your Flask application into:

app.py: The entry point to run the application.
config.py: Contains all the configuration settings.
models.py: Defines the database models.
views.py: Contains the route handlers (view functions).
extensions.py: Manages the initialization of Flask extensions like SQLAlchemy.
init.py: The app factory, responsible for creating and configuring the Flask application.

Debug School

How to perform crud in flask

Crud operation Using postman

Crud operation without postman

Full Code

Crud operation without postman

Top comments (0)