搜尋
首頁後端開發Python教學使用 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/<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')


</task_id>

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/<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')


</task_id>

這個 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 用於與MySQL 資料庫 連線和互動。當在映像建置過程中執行 pip install -rrequirements.txt 時,這些套件將安裝在 Docker 容器中。這確保應用程式擁有運行 Flask 伺服器 並與 MySQL 資料庫 通訊所需的工具。

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/<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')


</task_id>
建立所有檔案後,下一步是建置並執行服務,使用以下命令來建置和執行服務。


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/ 存取應用程式。

運行上述命令後,我能夠在瀏覽器上存取該應用程序,如下圖所示。

Task Manager App with Flask and MySQL

您可以使用 Postman 或 curl 來測試 /tasks 端點的 POST、GET 和 DELETE 操作。在這種情況下,我將使用curl。

捲曲命令:

  • 取得任務:
GET 方法取得所有任務。


docker-compose build
docker-compose up

Task Manager App with Flask and MySQL

請注意,每當您在瀏覽器上執行 http://localhost:5000/tasks 時,它都會顯示您已新增的所有任務,如新增任務中所述。

  • 新增任務:
POST 方法在資料庫中建立任務。


docker-compose up -d
這將向您的 Flask 應用程式發送帶有任務描述的 POST 請求。如果任務新增成功,您應該會收到以下回應:


docker-compose down
檢查瀏覽器的網路標籤或日誌以驗證 POST 請求是否正確發出。

我運行了該命令幾次,並自訂了「簡單任務」部分以產生不同的輸出,這是我運行的命令,輸出可以在下圖中看到。


docker ps 

Task Manager App with Flask and MySQL

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/<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')


</task_id>

Task Manager App with Flask and MySQL

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
);

Task Manager App with Flask and MySQL

  • 刪除任務:

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"]

Task Manager App with Flask and MySQL

結論

使用 Flask 和 MySQL 建立任務管理器應用程式是了解 Web 服務開發、資料庫整合和 Docker 容器化基礎知識的絕佳方法。

該專案封裝了 Web 伺服器和資料庫如何協同工作以提供無縫功能。

擁抱這種學習體驗,並將其用作更深層的網路和基於雲端的開發專案的墊腳石。

以上是使用 Flask 和 MySQL 的任務管理器應用程式的詳細內容。更多資訊請關注PHP中文網其他相關文章!

陳述
本文內容由網友自願投稿,版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容,請聯絡admin@php.cn
Numpy數組與使用數組模塊創建的數組有何不同?Numpy數組與使用數組模塊創建的數組有何不同?Apr 24, 2025 pm 03:53 PM

numpyArraysareAreBetterFornumericalialoperations andmulti-demensionaldata,而learthearrayModuleSutableforbasic,內存效率段

Numpy數組的使用與使用Python中的數組模塊陣列相比如何?Numpy數組的使用與使用Python中的數組模塊陣列相比如何?Apr 24, 2025 pm 03:49 PM

numpyArraySareAreBetterForHeAvyNumericalComputing,而lelethearRayModulesiutable-usemoblemory-connerage-inderabledsswithSimpleDatateTypes.1)NumpyArsofferVerverVerverVerverVersAtility andPerformanceForlargedForlargedAtatasetSetsAtsAndAtasEndCompleXoper.2)

CTYPES模塊與Python中的數組有何關係?CTYPES模塊與Python中的數組有何關係?Apr 24, 2025 pm 03:45 PM

ctypesallowscreatingingangandmanipulatingc-stylarraysinpython.1)usectypestoInterfacewithClibrariesForperfermance.2)createc-stylec-stylec-stylarraysfornumericalcomputations.3)passarraystocfunctions foreforfunctionsforeffortions.however.however,However,HoweverofiousofmemoryManageManiverage,Pressiveo,Pressivero

在Python的上下文中定義'數組”和'列表”。在Python的上下文中定義'數組”和'列表”。Apr 24, 2025 pm 03:41 PM

Inpython,一個“列表” isaversatile,mutableSequencethatCanholdMixedDatateTypes,而“陣列” isamorememory-sepersequeSequeSequeSequeSequeRingequiringElements.1)列表

Python列表是可變還是不變的?那Python陣列呢?Python列表是可變還是不變的?那Python陣列呢?Apr 24, 2025 pm 03:37 PM

pythonlistsandArraysareBothable.1)列表Sareflexibleandsupportereceneousdatabutarelessmory-Memory-Empefficity.2)ArraysareMoremoremoremoreMemoremorememorememorememoremorememogeneSdatabutlesserversEversementime,defteringcorcttypecrecttypececeDepeceDyusagetoagetoavoavoiDerrors。

Python vs. C:了解關鍵差異Python vs. C:了解關鍵差異Apr 21, 2025 am 12:18 AM

Python和C 各有優勢,選擇應基於項目需求。 1)Python適合快速開發和數據處理,因其簡潔語法和動態類型。 2)C 適用於高性能和系統編程,因其靜態類型和手動內存管理。

Python vs.C:您的項目選擇哪種語言?Python vs.C:您的項目選擇哪種語言?Apr 21, 2025 am 12:17 AM

選擇Python還是C 取決於項目需求:1)如果需要快速開發、數據處理和原型設計,選擇Python;2)如果需要高性能、低延遲和接近硬件的控制,選擇C 。

達到python目標:每天2小時的力量達到python目標:每天2小時的力量Apr 20, 2025 am 12:21 AM

通過每天投入2小時的Python學習,可以有效提升編程技能。 1.學習新知識:閱讀文檔或觀看教程。 2.實踐:編寫代碼和完成練習。 3.複習:鞏固所學內容。 4.項目實踐:應用所學於實際項目中。這樣的結構化學習計劃能幫助你係統掌握Python並實現職業目標。

See all articles

熱AI工具

Undresser.AI Undress

Undresser.AI Undress

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

AI Clothes Remover

AI Clothes Remover

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

Undress AI Tool

Undress AI Tool

免費脫衣圖片

Clothoff.io

Clothoff.io

AI脫衣器

Video Face Swap

Video Face Swap

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

熱工具

記事本++7.3.1

記事本++7.3.1

好用且免費的程式碼編輯器

Atom編輯器mac版下載

Atom編輯器mac版下載

最受歡迎的的開源編輯器

MinGW - Minimalist GNU for Windows

MinGW - Minimalist GNU for Windows

這個專案正在遷移到osdn.net/projects/mingw的過程中,你可以繼續在那裡關注我們。 MinGW:GNU編譯器集合(GCC)的本機Windows移植版本,可自由分發的導入函式庫和用於建置本機Windows應用程式的頭檔;包括對MSVC執行時間的擴展,以支援C99功能。 MinGW的所有軟體都可以在64位元Windows平台上運作。

禪工作室 13.0.1

禪工作室 13.0.1

強大的PHP整合開發環境

WebStorm Mac版

WebStorm Mac版

好用的JavaScript開發工具