When the backend is almost completed in DB product development I wanted to improve the user-friendliness by preparing a driver in the form used as standard in python and java as a wrapper.
I haven't written java yet, so I need to study it, but first I will write a wrapper in python, study the specifications of the standard driver, and summarize the implementation flow.
--python Driver for Mysql --python db API specifications
The following is a summary of what I tried to implement while naive this time.
-** Our_Api ** class -** Connection ** class -** Cursor ** class -** Error ** class -** Type ** class
Code-wise, using our driver
main.py
#!/usr/bin/python
import MySQLdb
# Open database connection
db = MySQLdb.connect("localhost","testuser","test123","TESTDB" )
# prepare a cursor object using cursor() method
cursor = db.cursor()
# execute SQL query using execute() method.
cursor.execute("SELECT VERSION()")
# Fetch a single row using fetchone() method.
data = cursor.fetchone()
print "Database version : %s " % data
# disconnect from server
db.close()
I want to do something like that.
Here, `MySQLdb``` is the library of the drivers we provide, ```mysqldb```But```connection```Create an object of the class with the required configuration variables. Also, the `Connectionclass creates an object of the Cursor``` class.
The cursor class is cursor.execute()Or cursor.fetchone()Has a method like
The passed SQL results are stored and you can access and view those results.
Also, when Insert or Update is performed,
#### **`main.py`**
```py
#!/usr/bin/python
import MySQLdb
# Open database connection
db = MySQLdb.connect("localhost","testuser","test123","TESTDB" )
# prepare a cursor object using cursor() method
cursor = db.cursor()
# Prepare SQL query to INSERT a record into the database.
sql = """INSERT INTO EMPLOYEE(FIRST_NAME,
LAST_NAME, AGE, SEX, INCOME)
VALUES ('Mac', 'Mohan', 20, 'M', 2000)"""
try:
# Execute the SQL command
cursor.execute(sql)
# Commit your changes in the database
db.commit()
except:
# Rollback in case there is any error
db.rollback()
# disconnect from server
db.close()
In this way, the changed operation is actually reflected in the DB by calling the `commit ()` method of the ``` db.coimmit () , that is, the connection``` class. If you want to get it back, you can do it with a method like `` db.rollback ()` ``.
Also, the Error class is implemented to handle errors in a specified way.
Furthermore, the Type class is used to accurately change the returned data type to the corresponding data type of python, such as when the backend of DB operation is written in other than Python. I will.
Based on the above, I will briefly introduce what I implemented.
In the main directory
tree .
The result of
.
├── api_core
│ ├── api_connection.py
│ ├── api_cursor.py
│ ├── api_error.py
│ ├── api_type.py
│ ├── __init__.py
│ └── our_api.py
├── main.py
└── requirements.txt
It is made up of.
api_connection.py
from .api_cursor import Cursor
class Connection:
'''
Connection class
class member:
api_host: str
api_port: int
e.g) http://{api_host}:{api_port}
class method:
cursor: returns Cursor class object
close: call destructor to delete the class object itself
'''
def __init__(self, proxy_conf_dict = None, proxy_conf_file=None):
if proxy_conf_dict:
self.api_host = proxy_conf_dict['api_host']
self.api_port = proxy_conf_dict['api_port']
self.proxy_conf_dict = proxy_conf_dict
def __del__(self):
pass
def cursor(self):
return Cursor(self.proxy_conf_dict)
def close(self):
self.__del__()
api_cursor.py
import requests
import json
from .api_type import Type, Type_Enum
class Cursor:
def __init__(self, proxy_conf_dict=None):
'''
Cursor class
class member:
result: list of retrieved records (list of dictionary)
type_dict: dict of (telling which column is datetime, time or date)
conf_dict: dict of (host and port)
index which: integer telling where cursor points to
query_execute_que: list of raw_query (query not related to select statement and they will be sent when connection.commit)
class method:
execute: params(raw_query:str)
update self.result and self.type_dict
commit: params()
send post request for "not" select statement
fetchone: params()
return self.result(self.index)
fetchall: params()
return self.result
next: params()
increment self.index by one
previous: params()
increment self.index by minus one
'''
# Cursor contains followings
self.result = None
self.type_dict = None
self.proxy_conf_dict = None
self.index = 0
self.query_execute_que = []
if conf_dict is not None:
self.conf_dict = conf_dict
def execute(self, raw_query:str):
# execute function
# sends query by post request to proxy, when sql is select statement
# if not select statement, store the sql to self.query_execute_que
if self.__parse_query_select_or_not(raw_query):
url = f'http://{self.conf_dict["api_host"]}:{self.conf_dict["api_port"]}/query'
result = requests.post(url, data=dict(sql=raw_query))
# post request to /query endpoint returns result (list of dictionary) and type_dict (dictionary)
self.result = json.loads(result.text)['result']
self.type_dict = json.loads(result.text)['type_dict']
# if type_dict contains key, mean that result contains either datetime, time or date
# therefore those records needs to be converted to the python class object instead of string.
if self.type_dict.keys() is not None and len(self.type_dict.keys()) > 0:
for i in range(len(self.result)):
for key in self.type_dict.keys():
self.result[i][key] = Type.parse_string_to(self.result[i][key], self.type_dict[key])
else:
self.query_execute_que.append(raw_query)
def commit(self):
# commit function
# if there are stored raw_query in self.query_execute_que, send post request
if len(self.query_execute_que) == 0:
pass
else:
url = f'http://{self.conf_dict["api_host"]}:{self.conf_dict["api_port"]}/query'
result = requests.post(url, data=dict(sql=self.query_execute_que, transaction=True))
self.query_execute_que = None
def fetchone(self):
# fetchone function
# if there is record (dictionary) in self.result,
# and if self.index is whithin the len(self.result)
# return result[self.index](one correspoinding record)
assert self.result is not None, 'cursor does not have a result to fetch'
if len(self.result) > 0:
try:
return self.result[self.index]
except:
raise Exception('cursor index is not appropriate for result')
else:
pass
pass
def fetchall(self):
# fetch all function
# if there is records (dictonary) in self.result,
# return all teh result (as a list of dictionary)
assert self.result is not None, 'cursor does not have a result to fetch'
if len(self.result) > 0:
return self.result
else:
pass
pass
def next(self):
# next function
# move index one forward
self.index += 1
def previous(self):
# previous function
# move index one backward
self.index -= 1
def __parse_query_select_or_not(self, raw_query:str):
# parser for raw_query
# raw_query: str
# return True if raw_query is select statement,
# False if raw_query is not select statement
if raw_query.lower().startswith('select'):
return True
else:
False
api_error.py
class Error:
'''
Error class
TODO implementation
'''
def __init__(self):
pass
api_type.py
from enum import Enum
import datetime
class Type_Enum(Enum):
'''
Type_Enum class
contains Data type which needs to be converted to python object from string in query result.
class member:
DATE = 'date'
DATETIME = 'datetime'
TIME = 'time'
'''
DATE = 'date'
DATETIME = 'datetime'
TIME = 'time'
class Type:
'''
Type class
class member:
class method:
@staticmethod
parse_string_to: params(target_string:str, target_type:str)
return either python datetime, date, time object which parsed string to.
'''
def __init__(self):
pass
## some data type such as date, datetime, time is returned as string, needs to be converted to corresponding python object.
@staticmethod
def parse_string_to(target_string:str, target_type:str):
if target_type == Type_Enum.DATE.value:
date_time_obj = datetime.datetime.strptime(target_string, '%Y-%m-%d')
return date_time_obj.date()
elif target_type == Type_Enum.TIME.value:
date_time_obj = datetime.datetime.strptime(target_string, '%H:%M:%S.%f')
return date_time_obj.time()
elif target_type == Type_Enum.DATETIME.value:
date_time_obj = datetime.datetime.strptime(target_string, '%Y-%m-%d %H:%M:%S.%f')
return date_time_obj
our_api.py
from .api_connection import Connection
class Our_API:
'''
Our_API class
class member:
class method:
connect: params(conf_dict, conf_file)
return Connection class object
'''
def __init__(self):
pass
def connect(conf_dict=None, conf_file=None):
if conf_dict:
return Connection(conf_dict)
elif conf_file:
return Connection(conf_file)
I'm looking forward to seeing what this looks like when written in Java.
end.
Recommended Posts