지역 사회 배우다 도구 라이브러리 여가
찾다
한국어
집 > 백엔드 개발 > 파이썬 튜토리얼 > 본문

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

Mary-Kate Olsen
풀어 주다： 2024-11-17 08:19:03
원래의
413명이 탐색했습니다.

프로젝트 개요

이 프로젝트는 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 중국어 웹사이트의 기타 관련 기사를 참조하세요!

원천：dev.to
이전 기사：`str.replace()`를 사용하여 Python 3.x에서 문자열 대체가 어떻게 변경되었습니까? 다음 기사：Python에서 사용자 입력 및 명령줄 인수를 어떻게 읽나요?
본 웹사이트의 성명
본 글의 내용은 네티즌들의 자발적인 기여로 작성되었으며, 저작권은 원저작자에게 있습니다. 본 사이트는 이에 상응하는 법적 책임을 지지 않습니다. 표절이나 침해가 의심되는 콘텐츠를 발견한 경우 admin@php.cn으로 문의하세요.
저자별 최신 기사
최신 이슈
function_exists()는 사용자 정의 함수를 결정할 수 없습니다. 기능 테스트() {                                                                                ...
에서 2024-04-29 11:01:01
0
2
1870
Chrome 모바일 버전을 표시하는 방법 안녕하세요 선생님, Chrome을 모바일 버전으로 어떻게 변경하나요?
에서 2024-04-23 00:22:19
0
10
2037
자식 창이 부모 창을 작동하지만 출력이 응답하지 않습니다. 처음 두 문장은 실행 가능하지만 마지막 문장은 구현할 수 없습니다.
에서 2024-04-19 15:37:47
0
1
1713
상위 창에 출력이 없습니다. document.onclick = function(){ window.opener.document.write('나는 자식 창의 출력입니다.');
에서 2024-04-18 23:52:34
0
1
1614
CSS 마인드맵 코스웨어는 어디에 있나요? 코스웨어
에서 2024-04-16 10:10:18
0
0
1627
관련 주제
더>
인기 추천
인기 튜토리얼
더>
관련 튜토리얼
인기 추천
최신 강좌
최신 다운로드
더>
웹 효과
웹사이트 소스 코드
웹사이트 자료
프론트엔드 템플릿