ホームページ >バックエンド開発 >Python チュートリアル >Flask と MySQL を使用したタスクマネージャーアプリ

Flask と MySQL を使用したタスクマネージャーアプリ

Mary-Kate Olsenオリジナル: 2024-11-17 08:19:03470ブラウズ

プロジェクト概要

このプロジェクトは、Flask と MySQL で構築された タスクマネージャーアプリ です。タスクを管理するためのシンプルな RESTful API を提供し、基本的な CRUD (作成、読み取り、削除) 操作を示します。

このアプリケーションは、Flask アプリケーションを Docker を使用してコンテナ化し、MySQL データベースに接続する方法を理解するのに最適です。

特徴

新しいタスクを追加
すべてのタスクを表示
ID でタスクを削除

Flask コード: 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
);

ファイルを保存します。

docker-compose.yml が配置されているプロジェクトフォルダーに、ファイルを init-db.sql として保存しました。 .

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 イメージを使用します。
- environment: 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 イメージをビルドします。
- ports: Flask アプリのポート 5000 をホストのポート 5000 にマッピングします。
- depends_on: db サービスが Web サービスより前に開始されるようにします。
- environment: Flask の環境変数を設定します。
- volumes: 現在のプロジェクトディレクトリをコンテナ内の /app ディレクトリにマウントします。 ### ボリュームセクション: db_data: コンテナーの再起動間で MySQL データを保持するための名前付きボリューム db_data を定義します。

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 ランタイムに python:3.9-slim を使用します。 作業ディレクトリ: /app を作業ディレクトリとして設定します。

2.依存関係:requirements.txt をコピーし、pip 経由で依存関係をインストールします。

3.ツールのインストール: サービスの準備状況を確認するために wait-for-it をインストールします。

4.アプリケーションコード: すべてのアプリコードをコンテナーにコピーします。

5.起動コマンド: app.py を起動する前に wait-for-it を実行して、MySQL DB (db:3306) の準備が整っていることを確認します。

Requirements.txt ファイル

この requirements.txt は、Python プロジェクトが Web アプリケーションの構築に Flask フレームワーク と mysql-connector-python を必要とすることを指定します。 との接続と対話用MySQL データベース。これらのパッケージは、イメージのビルドプロセス中に pip install -r required.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/<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/ でアプリにアクセスします。
下の画像に示すように、上記のコマンドを実行すると、ブラウザーでアプリにアクセスできました。

Task Manager App with Flask and MySQL

Postman または Curl を使用して、POST、GET、および DELETE 操作の /tasks エンドポイントをテストできます。この場合、curl を使用します。

カールコマンド:

タスクの取得:

GET メソッドはすべてのタスクを取得します。

docker-compose build
docker-compose up

Task Manager App with Flask and MySQL

ブラウザで http://localhost:5000/tasks を実行すると、タスクの追加で説明したように、追加したすべてのタスクが表示されることに注意してください。

タスクの追加:

POST メソッドはデータベース内にタスクを作成します。

docker-compose up -d

これにより、タスクの説明を含む POST リクエストが Flask アプリに送信されます。タスクが正常に追加されると、次のような応答を受け取るはずです:

docker-compose down

ブラウザのネットワークタブまたはログをチェックして、POST リクエストが正しく行われていることを確認してください。

コマンドを数回実行し、「Simple Task」と表示されている部分をカスタマイズして、さまざまな出力を生成しました。実行したコマンドは次のとおりです。出力は以下の画像で確認できます。

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

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 サーバーとデータベースが連携してシームレスな機能を提供する方法をカプセル化しています。

この学習経験を活用し、より深い Web およびクラウドベースの開発プロジェクトへの足がかりとして使用してください。

以上がFlask と MySQL を使用したタスクマネージャーアプリの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。

Python sql mysql flask restful postman pip define if for cURL Directory using var delete this table docker database http Access

声明：

この記事の内容はネチズンが自主的に寄稿したものであり、著作権は原著者に帰属します。このサイトは、それに相当する法的責任を負いません。盗作または侵害の疑いのあるコンテンツを見つけた場合は、admin@php.cn までご連絡ください。

前の記事：`str.replace()` を使用した Python 3.x の文字列置換はどう変わりましたか?次の記事：`str.replace()` を使用した Python 3.x の文字列置換はどう変わりましたか?

続きを見る

Flask と MySQL を使用したタスク マネージャー アプリ

プロジェクト概要

特徴

Flask コード: app.py

MySQL データベースのセットアップ スクリプト

ファイルを保存します。

Docker の構成

Requirements.txt ファイル

プロジェクトのテスト

カールコマンド:

結論

関連記事

Flask と MySQL を使用したタスクマネージャーアプリ

MySQL データベースのセットアップスクリプト