For my current phase project, I've chosen to develop a Personal Task Management App. I'll use it to explain, step by step, how to set up class resources, Tasks and TaskById, using Flask. These classes will handle creating, retrieving, updating, and deleting tasks.
Before you begin, make sure Python is installed in your system. Then, head to the terminal in VSCode and run the code below in the root folder:
pipenv install flask flask-sqlalchemy flask-restful
Step 1: Set Up Your Flask Application
We'll create a new Flask app and configure the necessary components:
from flask import Flask, request, session, make_response
from flask_sqlalchemy import SQLAlchemy
from flask_cors import CORS
from flask_migrate import Migrate
from flask_restful import Api, Resource
from datetime import datetime
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///data.db'
app.config['SQLALCHEMY_TRACK_MODIFICATION'] = False
app.config['SECRET_KEY'] = 'your_secret_key'
db = SQLAlchemy()
migrate = Migrate(app, db)
db.init_app(app)
api = Api(app)
CORS(app)
This step involves setting up the basic configuration for your Flask application, including integrating SQLAlchemy for database management and Flask-RESTful for building RESTful APIs.
-
Flask: The main framework to create web apps -
request: To handle incoming HTTP requests -
session: To manage user sessions (e.g., storing user_id when a user is logged in). -
make_response: To create HTTP responses. -
SQLAlchemy: A SQL toolkit and Object-Relational Mapping (ORM) system. -
CORS: Cross-Origin Resource Sharing (CORS), restricts web pages from making requests to a different domain than the one that served the web page. -
Migrate: To handle database migrations in a Flask application -
Api, Resource: Provided by Flask-RESTful, used to create RESTful API endpoints. -
datetime: To handle date and time-related tasks
Step 2: Define Task Model
Next, define the Task model, which represents a task in your database:
class Task(db.Model):
__tablename__ = "tasks"
id = db.Column(db.Integer, primary_key=True)
task_name = db.Column(db.String(255), nullable=False)
category = db.Column(db.String(255))
task_due_date = db.Column(db.DateTime)
task_status = db.Column(db.String, default="pending")
Step 3: Create the Tasks and TaskById Resource
- The
Tasksresource will handle requests to the/tasksendpoint. It will allow users to retrieve and create tasks. - The
TaskByIdresource will handle requests to the/tasks/<int:id>endpoint. It will allow users to retrieve, update, and delete a specific task.
class Tasks(Resource):
# /tasks
def get(self):
if 'user_id' in session:
my_tasks = Task.query.filter(Task.user_id == session['user_id']).all()
if len(my_tasks) > 0:
tasks = [task.to_dict() for task in my_tasks]
return make_response(tasks, 200)
return make_response({"message": "No task created yet"}, 200)
return make_response({"error": "Pls log in to view tasks."}, 401)
def post(self):
task_name = request.json.get('task_name')
category = request.json.get('category')
task_due_date = datetime.strptime(request.json.get('task_due_date'), '%Y-%m-%d')
task_status = request.json.get('task_status', 'pending')
new_task = Task(task_name=task_name, category=category, task_due_date=task_due_date, task_status=task_status, user_id=session['user_id'])
db.session.add(new_task)
db.session.commit()
if new_task.id:
return make_response(new_task.to_dict(), 201)
return make_response({"error": "error occurred"}, 400)
class TaskById(Resource):
# /tasks/<int:id>
@classmethod
def find_task(cls, id):
return Task.query.filter(and_(Task.id == id, Task.user_id == session['user_id'])).first()
def get(self, id):
if 'user_id' in session:
task = TaskById.find_task(id)
if task:
return make_response(task.to_dict(), 200)
return make_response({"error": "This task doesn't exist or you may not have permission to view this task"}, 401)
def patch(self, id):
task = TaskById.find_task(id)
if task:
for attr in request.json:
if attr == "task_due_date":
request.json[attr] = datetime.strptime(request.json[attr], '%Y-%m-%d')
setattr(task, attr, request.json[attr])
db.session.commit()
return make_response(task.to_dict(), 200)
return make_response({"error": "No task found"}, 404)
def delete(self, id):
task = TaskById.find_task(id)
if task:
db.session.delete(task)
db.session.commit()
return make_response({"message": "Task is deleted successfully"}, 200)
return make_response({"error": "No task found"}, 404)
Step 4: Add Resources to the API
api.add_resource(Tasks, '/tasks')
api.add_resource(TaskById, '/tasks/<int:id>')
-
db: An instance of SQLAlchemy, initialized with the Flask application instance. This sets up the connection between the Flask app and the database, allowing you to define and interact with database models. -
api: An instance of Flask-RESTful's Api. This sets up the framework for adding RESTful API resources to your application.
Top comments (0)