Lorsque le backend est presque terminé dans le développement du produit DB Je voulais améliorer la convivialité en préparant un pilote qui est utilisé en standard en python et java comme wrapper.
Je n'ai pas encore écrit java, donc je dois l'étudier, mais d'abord j'écrirai un wrapper en python, étudierai les spécifications du pilote standard et résumerai le flux d'implémentation.
--python Pilote pour Mysql --python spécifications de l'API db
Ce qui suit est un résumé de ce que j'ai essayé de mettre en œuvre tout en étant naïf cette fois.
Au niveau du code, en utilisant notre pilote
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()
Je veux faire quelque chose comme ça.
Ici, MySQLdb '' est la bibliothèque des pilotes que nous fournissons, ```mysqldb```Mais```connection```Créez un objet de la classe avec les variables de configuration requises. De plus, la classe Connection '' crée un objet de la classe `` Cursor ''.
La classe Cursor est le curseur.execute()Ou curseur.fetchone()A une méthode comme
Les résultats du SQL passé sont stockés et vous pouvez accéder et afficher ces résultats.
De plus, lors de l'insertion ou de la mise à jour,
#### **`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()
En appelant la méthode `commit ()` `de la` db.coimmit () '' c'est-à-dire la classe `connection```, l'opération modifiée est en fait reflétée dans la base de données. , Si vous souhaitez revenir en arrière, vous pouvez revenir en arrière avec une méthode comme `` db.rollback () ''.
De plus, la classe Error est implémentée pour gérer les erreurs d'une manière spécifiée.
De plus, la classe `` Type '' est utilisée pour changer avec précision le type de données retourné en type de données correspondant de python, comme lorsque le backend de l'opération DB est écrit dans un autre que Python. Je vais.
Sur la base de ce qui précède, je présenterai brièvement ce que j'ai mis en œuvre.
Dans le répertoire principal
tree .
Le résultat de
.
├── api_core
│ ├── api_connection.py
│ ├── api_cursor.py
│ ├── api_error.py
│ ├── api_type.py
│ ├── __init__.py
│ └── our_api.py
├── main.py
└── requirements.txt
Il est constitué de.
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)
J'ai hâte de voir à quoi cela ressemble une fois écrit en Java.
fin.