Apl Pengurus Tugas dengan Flask dan MySQL

Gambaran Keseluruhan Projek

Projek ini ialah Apl Pengurus Tugas yang dibina dengan Flask dan MySQL. Ia menyediakan API RESTful yang mudah untuk mengurus tugas, menunjukkan operasi asas CRUD (Buat, Baca, Padam).

Aplikasi ini sesuai untuk memahami cara aplikasi Flask boleh disimpan dalam bekas menggunakan Docker dan disambungkan dengan pangkalan data MySQL.

Ciri-ciri

• Tambah tugasan baharu
• Lihat semua tugasan
• Padamkan tugas dengan ID

Kod Kelalang: 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')

Skrip Persediaan Pangkalan Data MySQL

Buat skrip MySQL bernama init-db.sql untuk menyediakan pangkalan data dan jadual tugas:

Untuk mencipta skrip init-db.sql, ikut langkah berikut:

Buat fail baharu dalam direktori projek anda:

Navigasi ke folder projek dan buat fail baharu bernama init-db.sql
Tambahkan perintah SQL untuk menyediakan pangkalan data dan jadual tugas:

Buka init-db.sql dalam editor teks dan tambah arahan SQL berikut:

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

Simpan fail:

Saya menyimpan fail sebagai init-db.sql dalam folder projek di mana docker-compose.yml saya terletak .

Dalam docker-compose.yml:

Dalam fail docker-compose.yml saya, saya mempunyai konfigurasi volum yang menghala ke skrip ini.

Di bawah ialah fail docker-compose.yml

Konfigurasi 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:

Konfigurasi ini memastikan bahawa apabila bekas MySQL bermula, ia akan melaksanakan skrip init-db.sql untuk menyediakan task_db pangkalan data dan cipta jadual tugas.

Nota: Direktori docker-entrypoint-initdb.d/ digunakan oleh MySQL bekas untuk melaksanakan .sql skrip semasa permulaan awal bekas itu.

Penjelasan:

1. version: '3': Menentukan versi Docker Compose yang digunakan.

2. perkhidmatan:

• db:

  • imej: mysql:5.7: Menggunakan imej MySQL 5.7.
  • persekitaran: Menetapkan pembolehubah persekitaran untuk bekas MySQL:
    • MYSQL_ROOT_PASSWORD: Kata laluan akar untuk MySQL.
    • MYSQL_DATABASE: Pangkalan data yang akan dibuat semasa permulaan.
  • port: Petakan port kontena MySQL 3306 ke port 3306 hos anda.
  • jilid:
    • db_data:/var/lib/mysql: Mengekalkan data pangkalan data dalam volum Docker yang dipanggil db_data.
    • ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql: Melekapkan init-db.sql skrip ke dalam direktori permulaan bekas MYSQL supaya ia berjalan apabila bekas bermula.

• web:

  • bina: .: Membina imej Docker untuk apl Flask anda menggunakan Fail Docker dalam direktori semasa.
  • port: Petakan port 5000 apl Flask ke port 5000 hos anda.
  • bergantung_pada: Memastikan perkhidmatan db bermula sebelum perkhidmatan web.
  • persekitaran: Menetapkan pembolehubah persekitaran untuk Flask.
  • volum: Melekapkan direktori projek semasa ke dalam direktori /app di dalam bekas. ### bahagian jilid: db_data: Mentakrifkan db_data volum bernama untuk mengekalkan data MySQL antara bekas mula semula.

Fail Docker:

Tentukan arahan binaan untuk apl 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')

Fail Docker ini menyediakan persekitaran Python yang ringan untuk aplikasi Flask:

1. Imej Asas: Menggunakan python:3.9-slim untuk masa jalan Python yang minimum.
Direktori Kerja: Tetapkan /apl sebagai direktori kerja.

2. Dependencies: Menyalin requirements.txt dan memasang dependencies melalui pip.

3. Pemasangan Alat: Memasang tunggu-untuk-itu untuk menyemak kesediaan perkhidmatan.

4. Kod Aplikasi: Menyalin semua kod apl ke dalam bekas.

5. Perintah Permulaan: Menjalankan tunggu-untuk-itu untuk memastikan DB MySQL (db:3306) sedia sebelum memulakan app.py.

Fail Requirements.txt

Ini requirements.txt menyatakan bahawa projek Python memerlukan Flask framework untuk membina aplikasi web dan mysql-connector-python untuk menyambung dan berinteraksi dengan pangkalan data MySQL. Pakej ini akan dipasang dalam bekas Docker apabila pip install -r requirements.txt dijalankan semasa proses binaan imej. Ini memastikan apl mempunyai alatan yang diperlukan untuk menjalankan pelayan Flask dan berkomunikasi dengan pangkalan data 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/<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')

Selepas membuat semua fail, langkah seterusnya ialah membina dan menjalankan perkhidmatan arahan berikut digunakan untuk membina dan menjalankan perkhidmatan.

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

untuk menjalankan perkhidmatan dalam mod terpisah saya menggunakan arahan berikut dan bukannya 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:

apabila saya ingin menghentikan perkhidmatan saya menggunakan arahan

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

Sekarang setelah perkhidmatan dalam keadaan berjalan jalankan arahan

Flask
mysql-connector-python

untuk memastikan bekas berjalan

Kini tiba masanya untuk menyemak API perkhidmatan untuk memastikan ia berfungsi seperti yang diharapkan.

Menguji Projek

Akses apl di http://localhost:5000/ .
Saya dapat mengakses apl pada penyemak imbas saya selepas menjalankan arahan di atas seperti yang dilihat dalam imej di bawah.

Anda boleh menggunakan Posmen atau curl untuk menguji titik akhir /tasks untuk operasi POST, GET dan DELETE. Dalam kes ini saya akan menggunakan curl.

Arahan curl:

• Dapatkan Tugasan:

Kaedah GET mengambil semua tugasan.

docker-compose build
docker-compose up

Perhatikan bahawa pada bila-bila masa anda menjalankan http://localhost:5000/tasks pada penyemak imbas anda, ia menunjukkan kepada anda semua tugas yang telah anda tambahkan seperti yang dijelaskan dalam tugas tambah.

• Tambah Tugasan:

Kaedah POST mencipta tugasan dalam pangkalan data.

docker-compose up -d

Ini akan menghantar permintaan POST ke apl Flask anda dengan penerangan tugas. Jika tugasan berjaya ditambahkan, anda seharusnya menerima respons seperti:

docker-compose down

semak tab rangkaian penyemak imbas anda atau log untuk mengesahkan bahawa permintaan POST dibuat dengan betul.

Saya menjalankan arahan beberapa kali dan menyesuaikan bahagian yang dikatakan Tugas Mudah untuk menjana output yang berbeza di sini ialah arahan yang saya jalankan dan out put boleh dilihat dalam imej di bawah.

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

• Padamkan Tugasan:

Kaedah DELETE mengalih keluar tugas mengikut 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:

Saya menjalankan arahan di bawah untuk mengalih keluar tugas dengan ID:4 seperti yang dilihat dalam imej di bawah tugas 4 telah dialih keluar.

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

Kesimpulan

Mencipta Apl Pengurus Tugas menggunakan Flask dan MySQL ialah cara terbaik untuk memahami asas pembangunan perkhidmatan web, penyepaduan pangkalan data dan kontena dengan Docker.

Projek ini merangkum cara pelayan web dan pangkalan data berfungsi secara serentak untuk menyediakan kefungsian yang lancar.

Harapi pengalaman pembelajaran ini dan gunakannya sebagai batu loncatan kepada projek pembangunan web dan berasaskan awan yang lebih mendalam.

Atas ialah kandungan terperinci Apl Pengurus Tugas dengan Flask dan MySQL. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!