pip install pymysql

version: "3.9"

include:
  - "composes/mysql.yaml"

volumes:
  bdo-data: {}
  bdo-log: {}

networks:
  bdo-net:
    driver: bridge
    external: false

services:
  bdo:
    image: mysql:latest
    restart: always
    volumes:
      - type: volume
        source: "bdo-data"
        target: "/var/lib/mysql"
      - type: volume
        source: "bdo-log"
        target: "/var/log/mysql"
    ports:
      - "9905:3306"
    environment:
      MYSQL_ROOT_PASSWORD: bdo
      MYSQL_USER: bdo
      MYSQL_PASSWORD: bdo
      MYSQL_DATABASE: bdo
    container_name: bdo_db

docker-compose up -d --build

docker-compose ps

NAME                IMAGE               COMMAND                  SERVICE             CREATED             STATUS              PORTS
bdo_db              mysql:latest        "docker-entrypoint.s…"   bdo                 55 minutes ago      Up 55 minutes       33060/tcp, 0.0.0.0:9905->3306/tcp, :::9905->3306/tcp

docker exec -it bdo_db /bin/bash
mysql -u bdo -p

CREATE TABLE bdoTBL (
    id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
    boardNo INT,
    title VARCHAR(255)
) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

import configparser

properties: configparser.ConfigParser = configparser.ConfigParser()
properties.read('config/config.ini')

# get section
mysql: configparser.SectionProxy   = properties['MYSQL']

# get variables
mysql_port: int = int(mysql['PORT'])
mysql_username: str = mysql['USERNAME']
mysql_password: str = mysql['PASSWORD']
mysql_database: str = mysql['DATABASE']
mysql_host: str = mysql['HOST']

from datetime import datetime
import config.config as conf
import pymysql

class CRUD():
    def __init__(self) -> None:
        self.port: int = conf.mysql_port
        self.username: str = conf.mysql_username
        self.password: str = conf.mysql_password
        self.database: str = conf.mysql_database
        self.host: str = conf.mysql_host
    
    def date_time(self) -> str:
        return datetime.now().strftime('%Y-%m-%d %H:%M')
    
    """ 데이터베이스 연결 설정 """
    def connect_to_database(self) -> None:
        try:
            conn = pymysql.connect(
                host=self.host,
                port=self.port,
                user=self.username,
                password=self.password,
                database=self.database,
                charset='utf8mb4',
                cursorclass=pymysql.cursors.DictCursor) # 결과를 사전 방식으로 변환하도록 설정.
            print(f'[🟢 DB connected] {self.date_time()}')
            self.conn: pymysql.Connection = conn

        except pymysql.Error as e:
            print(f'Error {e.args[0]}: {e.args[1]}')

    """ Create"""
    def insert_data(self,table_name: str, data: tuple[str,str]) -> None:
        try:
            with self.conn.cursor() as cursor:
                # 중복 데이터를 필터링하기 위해 먼저 DB에서 검색.
                sql_select: str = f"SELECT * FROM {table_name} WHERE boardNo = %s"
                cursor.execute(sql_select,data[0])
                existing_data = cursor.fetchone()

                # 데이터베이스에 해당 데이터가 없으면 삽입
                if not existing_data:
                    # 데이터 삽입
                    sql_insert: str = f"INSERT INTO {table_name} (boardNo, title) VALUES (%s, %s)"
                    cursor.execute(sql_insert,data)

                    # 변경사항 데이터베이스에 반영
                    self.conn.commit()

                    print('[🟢 INSERT] 데이터 삽입 성공!')
                else:
                    print('[🔴 INSERT] 데이터 삽입 실패! (중복)')
        finally:
            pass

    """ Read """
    def select_data(self, table_name: str) -> None:
        try:
            with self.conn.cursor() as cursor:
                # 데이터 조회
                sql_select = f"SELECT * FROM {table_name}"
                cursor.execute(sql_select)
                result = cursor.fetchall()
                return result
        finally:
            pass

    """ Update """
    def update_data(self, table_name: str, new_value: str, condition_value: str) -> None:
        try:
            with self.conn.cursor() as cursor:
                # 데이터 업데이트
                sql_update: str = f"UPDATE {table_name} SET boardNo = %s WHERE boardNo = %s"
                cursor.execute(sql_update,(new_value, condition_value))

                # 변경사항 데이터베이스에 반영
                self.conn.commit()
                print('[🟢 UPDATE] 데이터 수정 성공!')
        finally:
            pass

    """ Delete """
    def delete_data(self, table_name: str, condition_value: str) -> None:
        try:
            with self.conn.cursor() as cursor:
                # 데이터 삭제
                sql_delete: str = f"DELETE FROM {table_name} WHERE boardNo = %s"
                cursor.execute(sql_delete, (condition_value,))

                # 변경사항 데이터베이스에 반영
                self.conn.commit()
                print('[🟢 UPDATE] 데이터 삭제 성공')
        finally:
            pass

def main() -> None:
    # Set db table name
    table_name: str = 'bdoTBL'

    # Create CRUD instance
    crud: CRUD = CRUD()

    # db 연결
    crud.connect_to_database()

    # 데이터 삽입
    data_to_insert: tuple[str,str] = (1, '새로운 업데이트 (v.0.0.1)')
    crud.insert_data(table_name=table_name,data=data_to_insert)

    # 데이터 조회
    select_data = crud.select_data(table_name=table_name)

    # 데이터 업데이트
    new_value, condition_value = (100, 1)
    crud.update_data(table_name=table_name,new_value=new_value, condition_value=condition_value)

    # 데이터 삭제
    condition_value: int = 100
    crud.delete_data(table_name=table_name,condition_value=condition_value)

if __name__ == '__main__':
    main()

python3 main.py