使用 Flask 和 MySQL 的任務管理器應用程式
項目概況
這個專案是一個使用 Flask 和 MySQL 建構的任務管理器應用程式。它提供了一個簡單的 RESTful API 來管理任務,演示了基本的 CRUD(建立、讀取、刪除)操作。
此應用程式非常適合了解如何使用 Docker 將 Flask 應用程式容器化並與 MySQL 資料庫連接。
特徵
- 新增任務
- 查看所有任務
- 透過ID刪除任務
燒瓶代碼:app.py
from flask import Flask, request, jsonify import mysql.connector from mysql.connector import Error app = Flask(__name__) # Database connection function def get_db_connection(): try: connection = mysql.connector.connect( host="db", user="root", password="example", database="task_db" ) return connection except Error as e: return str(e) # Route for the home page @app.route('/') def home(): return "Welcome to the Task Management API! Use /tasks to interact with tasks." # Route to create a new task @app.route('/tasks', methods=['POST']) def add_task(): task_description = request.json.get('description') if not task_description: return jsonify({"error": "Task description is required"}), 400 connection = get_db_connection() if isinstance(connection, str): # If connection fails return jsonify({"error": connection}), 500 cursor = connection.cursor() cursor.execute("INSERT INTO tasks (description) VALUES (%s)", (task_description,)) connection.commit() task_id = cursor.lastrowid cursor.close() connection.close() return jsonify({"message": "Task added successfully", "task_id": task_id}), 201 # Route to get all tasks @app.route('/tasks', methods=['GET']) def get_tasks(): connection = get_db_connection() if isinstance(connection, str): # If connection fails return jsonify({"error": connection}), 500 cursor = connection.cursor() cursor.execute("SELECT id, description FROM tasks") tasks = cursor.fetchall() cursor.close() connection.close() task_list = [{"id": task[0], "description": task[1]} for task in tasks] return jsonify(task_list), 200 # Route to delete a task by ID @app.route('/tasks/<int:task_id>', methods=['DELETE']) def delete_task(task_id): connection = get_db_connection() if isinstance(connection, str): # If connection fails return jsonify({"error": connection}), 500 cursor = connection.cursor() cursor.execute("DELETE FROM tasks WHERE id = %s", (task_id,)) connection.commit() cursor.close() connection.close() return jsonify({"message": "Task deleted successfully"}), 200 if __name__ == "__main__": app.run(host='0.0.0.0')
MySQL 資料庫設定腳本
建立名為 init-db.sql 的 MySQL 腳本來設定資料庫和任務表:
要建立 init-db.sql 腳本,請依照下列步驟操作:
在專案目錄中建立一個新的檔案:
導覽至專案資料夾並建立一個名為 init-db.sql
的新文件
新增 SQL 指令來設定資料庫和任務表:
在文字編輯器中開啟 init-db.sql 並新增下列 SQL 指令:
CREATE DATABASE IF NOT EXISTS task_db; USE task_db; CREATE TABLE IF NOT EXISTS tasks ( id INT AUTO_INCREMENT PRIMARY KEY, description VARCHAR(255) NOT NULL );
儲存文件:
我將檔案儲存為init-db.sql 位於我的docker-compose.yml 所在的專案資料夾中.
在 docker-compose.yml 中:
在我的 docker-compose.yml 檔案中,我有指向此腳本的捲配置。
下面是docker-compose.yml檔案
Docker 配置
docker-compose.yml:
version: '3' services: db: image: mysql:5.7 environment: MYSQL_ROOT_PASSWORD: example MYSQL_DATABASE: task_db ports: - "3306:3306" volumes: - db_data:/var/lib/mysql - ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql web: build: . ports: - "5000:5000" depends_on: - db environment: FLASK_ENV: development volumes: - .:/app volumes: db_data:
此配置確保當MySQL 容器啟動時,它將執行init-db.sql 腳本來設定task_db 資料庫並建立tasks 表。
注意:docker-entrypoint-initdb.d/目錄被MySQL容器用來執行.sql 容器初始啟動期間的腳本。 解釋:
1。 version: '3':
指定正在使用的 Docker Compose 版本。2。服務:
資料庫:
- image: mysql:5.7: 使用 MySQL 5.7 鏡像。
-
環境: 設定 MySQL 容器的環境變數:
- MYSQL_ROOT_PASSWORD: MySQL 的 root 密碼。
- MYSQL_DATABASE:啟動時所建立的資料庫。
- ports: 將 MySQL 容器的連接埠 3306 對應到主機的連接埠 3306。
-
卷:
- db_data:/var/lib/mysql: 將資料庫資料儲存在名為 db_data. 的 Docker 磁碟區中
- ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql: 掛載init-db.sql 腳本寫入MYSQL容器的初始化目錄,以便在容器啟動時運作。
-
網頁:
- build: .: 使用目前目錄中的 Dockerfile 為您的 Flask 應用建立 Docker 映像。
- 連接埠: 將 Flask 應用程式的連接埠 5000 對應到主機的連接埠 5000。
- depends_on: 確保 db 服務在 Web 服務之前啟動。
- 環境: 設定 Flask 的環境變數。
- volumes: 將目前專案目錄掛載到容器內的 /app 目錄中。 ### 卷部分: db_data: 定義一個命名卷 db_data 以在容器重新啟動之間保留 MySQL 資料。
Dockerfile:
定義 Flask 應用程式的建置指令:
from flask import Flask, request, jsonify import mysql.connector from mysql.connector import Error app = Flask(__name__) # Database connection function def get_db_connection(): try: connection = mysql.connector.connect( host="db", user="root", password="example", database="task_db" ) return connection except Error as e: return str(e) # Route for the home page @app.route('/') def home(): return "Welcome to the Task Management API! Use /tasks to interact with tasks." # Route to create a new task @app.route('/tasks', methods=['POST']) def add_task(): task_description = request.json.get('description') if not task_description: return jsonify({"error": "Task description is required"}), 400 connection = get_db_connection() if isinstance(connection, str): # If connection fails return jsonify({"error": connection}), 500 cursor = connection.cursor() cursor.execute("INSERT INTO tasks (description) VALUES (%s)", (task_description,)) connection.commit() task_id = cursor.lastrowid cursor.close() connection.close() return jsonify({"message": "Task added successfully", "task_id": task_id}), 201 # Route to get all tasks @app.route('/tasks', methods=['GET']) def get_tasks(): connection = get_db_connection() if isinstance(connection, str): # If connection fails return jsonify({"error": connection}), 500 cursor = connection.cursor() cursor.execute("SELECT id, description FROM tasks") tasks = cursor.fetchall() cursor.close() connection.close() task_list = [{"id": task[0], "description": task[1]} for task in tasks] return jsonify(task_list), 200 # Route to delete a task by ID @app.route('/tasks/<int:task_id>', methods=['DELETE']) def delete_task(task_id): connection = get_db_connection() if isinstance(connection, str): # If connection fails return jsonify({"error": connection}), 500 cursor = connection.cursor() cursor.execute("DELETE FROM tasks WHERE id = %s", (task_id,)) connection.commit() cursor.close() connection.close() return jsonify({"message": "Task deleted successfully"}), 200 if __name__ == "__main__": app.run(host='0.0.0.0')
這個 Dockerfile 為 Flask 應用程式設定了一個輕量級的 Python 環境:
1。基礎鏡像: 使用 python:3.9-slim 來實現最短的 Python 運行時間。
工作目錄:將 /app 設定為工作目錄。
2。依賴項:複製requirements.txt並透過pip安裝依賴項。
3。工具安裝: 安裝 wait-for-it 以檢查服務準備。
4。應用程式程式碼: 將所有應用程式程式碼複製到容器中。
5。啟動指令: 執行 wait-for-it 以確保 MySQL DB (db:3306) 在啟動 app.py 之前準備就緒。
需求.txt 檔案
此requirements.txt 指定Python 專案需要Flask 框架 來建立Web 應用程式和mysql-connector-python
from flask import Flask, request, jsonify import mysql.connector from mysql.connector import Error app = Flask(__name__) # Database connection function def get_db_connection(): try: connection = mysql.connector.connect( host="db", user="root", password="example", database="task_db" ) return connection except Error as e: return str(e) # Route for the home page @app.route('/') def home(): return "Welcome to the Task Management API! Use /tasks to interact with tasks." # Route to create a new task @app.route('/tasks', methods=['POST']) def add_task(): task_description = request.json.get('description') if not task_description: return jsonify({"error": "Task description is required"}), 400 connection = get_db_connection() if isinstance(connection, str): # If connection fails return jsonify({"error": connection}), 500 cursor = connection.cursor() cursor.execute("INSERT INTO tasks (description) VALUES (%s)", (task_description,)) connection.commit() task_id = cursor.lastrowid cursor.close() connection.close() return jsonify({"message": "Task added successfully", "task_id": task_id}), 201 # Route to get all tasks @app.route('/tasks', methods=['GET']) def get_tasks(): connection = get_db_connection() if isinstance(connection, str): # If connection fails return jsonify({"error": connection}), 500 cursor = connection.cursor() cursor.execute("SELECT id, description FROM tasks") tasks = cursor.fetchall() cursor.close() connection.close() task_list = [{"id": task[0], "description": task[1]} for task in tasks] return jsonify(task_list), 200 # Route to delete a task by ID @app.route('/tasks/<int:task_id>', methods=['DELETE']) def delete_task(task_id): connection = get_db_connection() if isinstance(connection, str): # If connection fails return jsonify({"error": connection}), 500 cursor = connection.cursor() cursor.execute("DELETE FROM tasks WHERE id = %s", (task_id,)) connection.commit() cursor.close() connection.close() return jsonify({"message": "Task deleted successfully"}), 200 if __name__ == "__main__": app.run(host='0.0.0.0')
CREATE DATABASE IF NOT EXISTS task_db; USE task_db; CREATE TABLE IF NOT EXISTS tasks ( id INT AUTO_INCREMENT PRIMARY KEY, description VARCHAR(255) NOT NULL );
docker-compose up
version: '3' services: db: image: mysql:5.7 environment: MYSQL_ROOT_PASSWORD: example MYSQL_DATABASE: task_db ports: - "3306:3306" volumes: - db_data:/var/lib/mysql - ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql web: build: . ports: - "5000:5000" depends_on: - db environment: FLASK_ENV: development volumes: - .:/app volumes: db_data:
FROM python:3.9-slim WORKDIR /app # Install dependencies COPY requirements.txt . RUN pip install -r requirements.txt # Install wait-for-it tool# RUN apt-get update && apt-get install -y wait-for-it #Copy the application code> COPY . . # Use wait-for-it to wait for DB and start the Flask app CMD ["wait-for-it", "db:3306", "--", "python", "app.py"]
Flask mysql-connector-python
現在是時候檢查服務 API 以確保它們按預期工作了。
測試項目
透過 http://localhost:5000/ 存取應用程式。
運行上述命令後,我能夠在瀏覽器上存取該應用程序,如下圖所示。
捲曲命令:
- 取得任務:
docker-compose build docker-compose up
- 新增任務:
docker-compose up -d
docker-compose down
我運行了該命令幾次,並自訂了「簡單任務」部分以產生不同的輸出,這是我運行的命令,輸出可以在下圖中看到。
docker ps
from flask import Flask, request, jsonify import mysql.connector from mysql.connector import Error app = Flask(__name__) # Database connection function def get_db_connection(): try: connection = mysql.connector.connect( host="db", user="root", password="example", database="task_db" ) return connection except Error as e: return str(e) # Route for the home page @app.route('/') def home(): return "Welcome to the Task Management API! Use /tasks to interact with tasks." # Route to create a new task @app.route('/tasks', methods=['POST']) def add_task(): task_description = request.json.get('description') if not task_description: return jsonify({"error": "Task description is required"}), 400 connection = get_db_connection() if isinstance(connection, str): # If connection fails return jsonify({"error": connection}), 500 cursor = connection.cursor() cursor.execute("INSERT INTO tasks (description) VALUES (%s)", (task_description,)) connection.commit() task_id = cursor.lastrowid cursor.close() connection.close() return jsonify({"message": "Task added successfully", "task_id": task_id}), 201 # Route to get all tasks @app.route('/tasks', methods=['GET']) def get_tasks(): connection = get_db_connection() if isinstance(connection, str): # If connection fails return jsonify({"error": connection}), 500 cursor = connection.cursor() cursor.execute("SELECT id, description FROM tasks") tasks = cursor.fetchall() cursor.close() connection.close() task_list = [{"id": task[0], "description": task[1]} for task in tasks] return jsonify(task_list), 200 # Route to delete a task by ID @app.route('/tasks/<int:task_id>', methods=['DELETE']) def delete_task(task_id): connection = get_db_connection() if isinstance(connection, str): # If connection fails return jsonify({"error": connection}), 500 cursor = connection.cursor() cursor.execute("DELETE FROM tasks WHERE id = %s", (task_id,)) connection.commit() cursor.close() connection.close() return jsonify({"message": "Task deleted successfully"}), 200 if __name__ == "__main__": app.run(host='0.0.0.0')
CREATE DATABASE IF NOT EXISTS task_db; USE task_db; CREATE TABLE IF NOT EXISTS tasks ( id INT AUTO_INCREMENT PRIMARY KEY, description VARCHAR(255) NOT NULL );
- 刪除任務:
DELETE 方法透過 ID 刪除任務。
version: '3' services: db: image: mysql:5.7 environment: MYSQL_ROOT_PASSWORD: example MYSQL_DATABASE: task_db ports: - "3306:3306" volumes: - db_data:/var/lib/mysql - ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql web: build: . ports: - "5000:5000" depends_on: - db environment: FLASK_ENV: development volumes: - .:/app volumes: db_data:
我執行了以下命令來刪除 ID 為 4 的任務,如下圖所示,任務 4 已被刪除。
FROM python:3.9-slim WORKDIR /app # Install dependencies COPY requirements.txt . RUN pip install -r requirements.txt # Install wait-for-it tool# RUN apt-get update && apt-get install -y wait-for-it #Copy the application code> COPY . . # Use wait-for-it to wait for DB and start the Flask app CMD ["wait-for-it", "db:3306", "--", "python", "app.py"]
結論
使用 Flask 和 MySQL 建立任務管理器應用程式是了解 Web 服務開發、資料庫整合和 Docker 容器化基礎知識的絕佳方法。
該專案封裝了 Web 伺服器和資料庫如何協同工作以提供無縫功能。
擁抱這種學習體驗,並將其用作更深層的網路和基於雲端的開發專案的墊腳石。
以上是使用 Flask 和 MySQL 的任務管理器應用程式的詳細內容。更多資訊請關注PHP中文網其他相關文章!

熱AI工具

Undresser.AI Undress
人工智慧驅動的應用程序,用於創建逼真的裸體照片

AI Clothes Remover
用於從照片中去除衣服的線上人工智慧工具。

Undress AI Tool
免費脫衣圖片

Clothoff.io
AI脫衣器

Video Face Swap
使用我們完全免費的人工智慧換臉工具,輕鬆在任何影片中換臉!

熱門文章

熱工具

記事本++7.3.1
好用且免費的程式碼編輯器

SublimeText3漢化版
中文版,非常好用

禪工作室 13.0.1
強大的PHP整合開發環境

Dreamweaver CS6
視覺化網頁開發工具

SublimeText3 Mac版
神級程式碼編輯軟體(SublimeText3)

Python适合数据科学、Web开发和自动化任务,而C 适用于系统编程、游戏开发和嵌入式系统。Python以简洁和强大的生态系统著称,C 则以高性能和底层控制能力闻名。

兩小時內可以學到Python的基礎知識。 1.學習變量和數據類型,2.掌握控制結構如if語句和循環,3.了解函數的定義和使用。這些將幫助你開始編寫簡單的Python程序。

Python在遊戲和GUI開發中表現出色。 1)遊戲開發使用Pygame,提供繪圖、音頻等功能,適合創建2D遊戲。 2)GUI開發可選擇Tkinter或PyQt,Tkinter簡單易用,PyQt功能豐富,適合專業開發。

2小時內可以學會Python的基本編程概念和技能。 1.學習變量和數據類型,2.掌握控制流(條件語句和循環),3.理解函數的定義和使用,4.通過簡單示例和代碼片段快速上手Python編程。

Python更易學且易用,C 則更強大但複雜。 1.Python語法簡潔,適合初學者,動態類型和自動內存管理使其易用,但可能導致運行時錯誤。 2.C 提供低級控制和高級特性,適合高性能應用,但學習門檻高,需手動管理內存和類型安全。

要在有限的時間內最大化學習Python的效率,可以使用Python的datetime、time和schedule模塊。 1.datetime模塊用於記錄和規劃學習時間。 2.time模塊幫助設置學習和休息時間。 3.schedule模塊自動化安排每週學習任務。

Python在web開發、數據科學、機器學習、自動化和腳本編寫等領域有廣泛應用。 1)在web開發中,Django和Flask框架簡化了開發過程。 2)數據科學和機器學習領域,NumPy、Pandas、Scikit-learn和TensorFlow庫提供了強大支持。 3)自動化和腳本編寫方面,Python適用於自動化測試和系統管理等任務。

Python在自動化、腳本編寫和任務管理中表現出色。 1)自動化:通過標準庫如os、shutil實現文件備份。 2)腳本編寫:使用psutil庫監控系統資源。 3)任務管理:利用schedule庫調度任務。 Python的易用性和豐富庫支持使其在這些領域中成為首選工具。
