LU06.L04 - Erstellen einer RESTful API für die ToDo-Liste mit Flask
main.py
- main.py
from flask import Flask, jsonify, request from todoItem import TodoItem # Stellen Sie sicher, dass die TodoItem-Klasse importiert ist from todoDao import TodoDao # Stellen Sie sicher, dass die TodoDao-Klasse importiert ist app = Flask(__name__) dao = TodoDao('todo_example.db') dao.create_table() @app.route('/todos', methods=['POST']) def add_todo(): data = request.get_json() new_item = TodoItem(None, data['title'], data['is_completed']) dao.add_item(new_item) return jsonify({"message": "Todo item created"}), 201 @app.route('/todos', methods=['GET']) def get_all_todos(): items = dao.get_all_items() return jsonify([item.__dict__ for item in items]), 200 @app.route('/todos/<int:item_id>', methods=['GET']) def get_todo(item_id): item = dao.get_item(item_id) if item: return jsonify(item.__dict__), 200 else: return jsonify({"message": "Item not found"}), 404 @app.route('/todos/<int:item_id>', methods=['PUT']) def update_todo(item_id): data = request.get_json() updated_item = TodoItem(item_id, data['title'], data['is_completed']) if dao.update_item(updated_item): return jsonify({"message": "Item updated"}), 200 else: return jsonify({"message": "Item not found or not updated"}), 404 @app.route('/todos/<int:item_id>', methods=['DELETE']) def delete_todo(item_id): if dao.delete_item(item_id): return jsonify({"message": "Item deleted"}), 200 else: return jsonify({"message": "Item not found or not deleted"}), 404 if __name__ == '__main__': app.run()
todoDao.py
- todoDao.py
import sqlite3 from todoItem import TodoItem # TODO: Implementiere die TodoDao-Klasse für CRUD-Operationen class TodoDao: def __init__(self, db_file): self.conn = sqlite3.connect(db_file, check_same_thread=False) self.cursor = self.conn.cursor() def create_table(self): self.cursor.execute('''DROP TABLE IF EXISTS todo_items''') self.cursor.execute( '''CREATE TABLE IF NOT EXISTS todo_items (item_id INTEGER PRIMARY KEY, title TEXT, is_completed BOOLEAN)''') self.conn.commit() def add_item(self, todo_item): self.cursor.execute("INSERT INTO todo_items (title, is_completed) VALUES (?, ?)", (todo_item.title, todo_item.is_completed)) self.conn.commit() def get_item(self, item_id): self.cursor.execute("SELECT * FROM todo_items WHERE item_id = ?", (item_id,)) row = self.cursor.fetchone() if row: return TodoItem(row[0], row[1], row[2]) return None def get_all_items(self): self.cursor.execute("SELECT * FROM todo_items") rows = self.cursor.fetchall() todo_items = [TodoItem(row[0], row[1], row[2]) for row in rows] return todo_items def update_item(self, todo_item): self.cursor.execute("UPDATE todo_items SET title = ?, is_completed = ? WHERE item_id = ?", (todo_item.title, todo_item.is_completed, todo_item.item_id)) if self.cursor.rowcount > 0: self.conn.commit() return True return False def delete_item(self, item_id): self.cursor.execute("DELETE FROM todo_items WHERE item_id = ?", (item_id,)) if self.cursor.rowcount > 0: self.conn.commit() return True return False def close(self): self.conn.close()
todoItem.py
- todoItem.py
from dataclasses import dataclass # TODO: Implementiere die TodoItem-Klasse mit @dataclass @dataclass class TodoItem: item_id: int title: str is_completed: bool