찾다

집  >  기사  >  백엔드 개발  >  Flask 및 MySQL을 사용한 작업 관리자 앱

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

Mary-Kate Olsen
Mary-Kate Olsen원래의
2024-11-17 08:19:03415검색

프로젝트 개요

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

파일을 저장합니다.

파일을 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/<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. 종속성: 요구사항.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/<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 또는 컬을 사용하여 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/<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을 사용하여 작업 관리자 앱을 만드는 것은 웹 서비스 개발, 데이터베이스 통합, Docker를 사용한 컨테이너화의 기본 사항을 이해하는 훌륭한 방법입니다.

이 프로젝트는 웹 서버와 데이터베이스가 조화롭게 작동하여 원활한 기능을 제공하는 방법을 요약합니다.

이 학습 경험을 받아들이고 이를 더 심층적인 웹 및 클라우드 기반 개발 프로젝트를 위한 디딤돌로 활용하세요.

위 내용은 Flask 및 MySQL을 사용한 작업 관리자 앱의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!

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에서 문자열 대체가 어떻게 변경되었습니까?

관련 기사

더보기