Notice
Recent Posts
Recent Comments
Link
일 | 월 | 화 | 수 | 목 | 금 | 토 |
---|---|---|---|---|---|---|
1 | 2 | |||||
3 | 4 | 5 | 6 | 7 | 8 | 9 |
10 | 11 | 12 | 13 | 14 | 15 | 16 |
17 | 18 | 19 | 20 | 21 | 22 | 23 |
24 | 25 | 26 | 27 | 28 | 29 | 30 |
Tags
- 라즈베리파이
- PER
- MS-SQL
- urllib
- pandas
- 다이어트
- Unity
- Excel
- node.js
- 유니티
- swift
- 날짜
- GIT
- MySQL
- PyQt
- sqlite
- javascript
- mssql
- ubuntu
- python
- IOS
- 맛집
- PyQt5
- tensorflow
- flutter
- 함수
- 리눅스
- ASP
- port
- Linux
Archives
아미(아름다운미소)
python mysql dbhelper 본문
import mysql.connector
from mysql.connector import Error
class DBHelper:
def __init__(self, host, user, password, database):
self.host = host
self.user = user
self.password = password
self.database = database
self.connection = None
self.cursor = None
def connect(self):
try:
self.connection = mysql.connector.connect(
host=self.host,
user=self.user,
password=self.password,
database=self.database
)
self.cursor = self.connection.cursor()
print("MySQL 데이터베이스 연결 성공")
except Error as e:
print(f"MySQL 데이터베이스 연결 실패: {e}")
def disconnect(self):
if self.connection.is_connected():
self.cursor.close()
self.connection.close()
print("MySQL 데이터베이스 연결 종료")
def execute_query(self, query, params=None):
try:
self.cursor.execute(query, params)
self.connection.commit()
return self.cursor
except Error as e:
print(f"쿼리 실행 실패: {e}")
self.connection.rollback()
return None
def create(self, table_name, data):
columns = ", ".join(data.keys())
values = ", ".join(["%s"] * len(data))
query = f"INSERT INTO {table_name} ({columns}) VALUES ({values})"
return self.execute_query(query, tuple(data.values()))
def read(self, table_name, conditions=None):
query = f"SELECT * FROM {table_name}"
if conditions:
query += " WHERE " + " AND ".join([f"{key}=%s" for key in conditions])
return self.execute_query(query, tuple(conditions.values()))
def update(self, table_name, data, conditions):
set_clause = ", ".join([f"{key}=%s" for key in data])
where_clause = " AND ".join([f"{key}=%s" for key in conditions])
query = f"UPDATE {table_name} SET {set_clause} WHERE {where_clause}"
return self.execute_query(query, tuple(data.values()) + tuple(conditions.values()))
def delete(self, table_name, conditions):
where_clause = " AND ".join([f"{key}=%s" for key in conditions])
query = f"DELETE FROM {table_name} WHERE {where_clause}"
return self.execute_query(query, tuple(conditions.values()))
예제:
# DBHelper 인스턴스 생성
db = DBHelper(host='localhost', user='your_username', password='your_password', database='your_database')
# 데이터베이스 연결
db.connect()
# 새 레코드 생성
new_user = {'name': 'John Doe', 'email': 'john.doe@example.com'}
db.create('users', new_user)
# 레코드 읽기
user = db.read('users', {'email': 'john.doe@example.com'}).fetchone()
print(user)
# 레코드 업데이트
updated_user = {'name': 'Jane Doe'}
db.update('users', updated_user, {'email': 'john.doe@example.com'})
# 레코드 삭제
db.delete('users', {'email': 'john.doe@example.com'})
# 데이터베이스 연결 종료
db.disconnect()
'랭귀지 > python' 카테고리의 다른 글
미체결 날짜형식을 현재시간과 비교해서 1분 경과시 주문취소 (0) | 2024.08.25 |
---|---|
flask restapi mysql 예제 (0) | 2024.06.22 |
flake8 설정파일 (0) | 2024.06.14 |
CSRF란? (0) | 2022.10.15 |
ORM (0) | 2022.10.12 |
Comments