▶︎ 개요
pymysql
라이브러리를 사용하여 mysql 서버에 접속해보고
간단한 CRUD 작업을 수행하기 위한 로직을 구현해보자.
▶︎ 라이브러리 설치
pip install pymysql
▶︎ MySQL 구축
- Docker를 사용하여 구축하도록 함.
docker-compose.yaml
version: "3.9"
include:
- "composes/mysql.yaml"
volumes:
bdo-data: {}
bdo-log: {}
networks:
bdo-net:
driver: bridge
external: false
composes/mysql.yaml
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
‣ 테이블 생성
먼저 bdo_db
컨테이너의 터미널로 접속 후, mysql 로그인
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;
▶︎ 코드 작성
config/config.py
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']
main.py
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