105 lines
2.6 KiB
Python
105 lines
2.6 KiB
Python
from flask import jsonify
|
|
from flask.views import MethodView
|
|
from sqlalchemy.orm import Session
|
|
from werkzeug.exceptions import NotFound
|
|
|
|
from tofu_api.common.rest import BaseBlueprint
|
|
from tofu_api.models import Task
|
|
|
|
|
|
class TaskApiBlueprint(BaseBlueprint):
|
|
"""
|
|
Blueprint for the tasks REST API.
|
|
"""
|
|
|
|
# Blueprint settings
|
|
name = 'rest_api_tasks'
|
|
import_name = __name__
|
|
url_prefix = '/tasks'
|
|
|
|
def init_blueprint(self) -> None:
|
|
"""
|
|
Register URL rules.
|
|
"""
|
|
db_session = self.app.dependencies.get_db_session()
|
|
|
|
self.add_url_rule(
|
|
'',
|
|
view_func=self.create_view_func(TaskCollectionView, db_session=db_session),
|
|
methods=['GET', 'POST'],
|
|
)
|
|
self.add_url_rule(
|
|
'/<int:task_id>',
|
|
view_func=self.create_view_func(TaskItemView, db_session=db_session),
|
|
methods=['GET', 'PATCH', 'DELETE'],
|
|
)
|
|
|
|
|
|
class TaskBaseView(MethodView):
|
|
"""
|
|
Base class for view classes for the `/tasks` endpoint.
|
|
"""
|
|
|
|
# TODO: Use a handler class instead of accessing the database session directly
|
|
db_session: Session
|
|
|
|
def __init__(self, *, db_session: Session):
|
|
self.db_session = db_session
|
|
|
|
|
|
class TaskCollectionView(TaskBaseView):
|
|
"""
|
|
View class for `/tasks` endpoint.
|
|
"""
|
|
|
|
def get(self):
|
|
"""
|
|
Get list of all tasks.
|
|
"""
|
|
task_list = self.db_session.query(Task).all()
|
|
return jsonify({
|
|
'count': len(task_list),
|
|
'items': [task.to_dict() for task in task_list],
|
|
}), 200
|
|
|
|
def post(self):
|
|
"""
|
|
Create a new task.
|
|
"""
|
|
# TODO: Parse request data and create real data
|
|
new_task = Task(
|
|
title='Do stuff'
|
|
)
|
|
self.db_session.add(new_task)
|
|
self.db_session.commit()
|
|
return jsonify(new_task.to_dict()), 201
|
|
|
|
|
|
class TaskItemView(TaskBaseView):
|
|
"""
|
|
View class for `/tasks/<int:task_id>` endpoint.
|
|
"""
|
|
|
|
def get(self, task_id: int):
|
|
"""
|
|
Get a single task by ID.
|
|
"""
|
|
task = self.db_session.query(Task).get(task_id)
|
|
if task is None:
|
|
raise NotFound(f'Task with ID {task_id} not found!')
|
|
return jsonify(task.to_dict()), 200
|
|
|
|
def patch(self, task_id: int):
|
|
"""
|
|
Update a single task by ID.
|
|
"""
|
|
# TODO: Implement
|
|
raise NotImplementedError
|
|
|
|
def delete(self, task_id: int):
|
|
"""
|
|
Delete a single task by ID.
|
|
"""
|
|
# TODO: Implement
|
|
raise NotImplementedError
|