41 lines
1.2 KiB
Python
41 lines
1.2 KiB
Python
import contextlib
|
|
|
|
import pymysql
|
|
from pymysql.constants import CLIENT
|
|
|
|
from constants import CONNECTION_TIMEOUT, DEFAULT_SCHEMA
|
|
|
|
|
|
class Database:
|
|
|
|
__connection: pymysql.Connection = None
|
|
__host: str = None
|
|
__user: str = None
|
|
__password: str = None
|
|
__database: str = None
|
|
|
|
def __init__(self, host: str, user: str, password: str) -> None:
|
|
self.__host = host
|
|
self.__user = user
|
|
self.__password = password
|
|
self.__database = DEFAULT_SCHEMA
|
|
|
|
def connect(self):
|
|
connection = pymysql.connect(host=self.__host, user=self.__user, passwd=self.__password,
|
|
database=self.__database, connect_timeout=CONNECTION_TIMEOUT,
|
|
client_flag=CLIENT.MULTI_STATEMENTS)
|
|
self.__connection = connection
|
|
|
|
@contextlib.contextmanager
|
|
def query(self, query: str):
|
|
if self.__connection is None:
|
|
raise Exception('データベースに接続されていません')
|
|
|
|
with self.__connection.cursor() as cursor:
|
|
cursor.execute(query)
|
|
yield cursor
|
|
|
|
def close(self):
|
|
self.__connection.close()
|
|
self.__connection = None
|