Flask 및 MySQL을 사용한 작업 관리자 앱-파이썬 튜토리얼-php.cn

집

백엔드 개발

파이썬 튜토리얼

Flask 및 MySQL을 사용한 작업 관리자 앱

Mary-Kate Olsen

Nov 17, 2024 am 08:19 AM

프로젝트 개요

이 프로젝트는 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
);

파일을 저장합니다.

파일을 docker-compose.yml이 있는 프로젝트 폴더에 init-db.sql으로 저장했습니다. .

docker-compose.yml에서:

내 docker-compose.yml 파일에 이 스크립트를 가리키는 볼륨 구성이 있습니다.

아래는 docker-compose.yml 파일입니다

도커 구성

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. 서비스:

db:
- 이미지: mysql:5.7: MySQL 5.7 이미지를 사용합니다.
- environment: MySQL 컨테이너에 대한 환경 변수를 설정합니다.
  - MYSQL_ROOT_PASSWORD: MySQL의 루트 비밀번호입니다.
  - 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에 매핑합니다.
- dependent_on: db 서비스가 웹 서비스보다 먼저 시작되도록 합니다.
- environment: Flask의 환경 변수를 설정합니다.
- volumes: 현재 프로젝트 디렉토리를 컨테이너 내부의 /app 디렉토리에 마운트합니다. ### 볼륨 섹션: db_data: 컨테이너가 다시 시작될 때까지 MySQL 데이터를 유지하기 위해 명명된 볼륨 db_data를 정의합니다.

도커파일:

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 런타임을 위해 python:3.9-slim을 사용합니다. 작업 디렉터리: /app를 작업 디렉터리로 설정합니다.

2. 종속성: 요구사항.txt를 복사하고 pip를 통해 종속성을 설치합니다.

3. 도구 설치: 서비스 준비 상태 확인을 위한 대기 기능을 설치합니다.

4. 애플리케이션 코드: 모든 앱 코드를 컨테이너에 복사합니다.

5. 시작 명령: app.py를 시작하기 전에 wait-for-it을 실행하여 MySQL DB(db:3306)가 준비되었는지 확인합니다.

요구사항.txt 파일

이 requirements.txt는 Python 프로젝트에 웹 애플리케이션 구축을 위한 Flask 프레임워크와 mysql-connector-python이 필요함을 지정합니다. MySQL 데이터베이스에 연결하고 상호 작용합니다. 이러한 패키지는 이미지 빌드 프로세스 중에 pip install -r 요구사항.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 또는 컬을 사용하여 POST, GET 및 DELETE 작업에 대한 /tasks 엔드포인트를 테스트할 수 있습니다. 이 경우에는 컬을 사용하게 됩니다.

컬 명령:

작업 가져오기:

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 요청이 올바르게 이루어지고 있는지 확인하세요.

명령어를 몇 번 실행하고 Simple Task to generate other Outputs라는 부분을 사용자 정의했습니다. 여기서 실행한 명령은 아래 이미지에서 확인할 수 있습니다.

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