Wenn das Backend in der DB-Produktentwicklung fast fertig ist Ich wollte die Benutzerfreundlichkeit verbessern, indem ich einen Treiber vorbereitete, der standardmäßig in Python und Java als Wrapper verwendet wird.
Ich habe Java noch nicht geschrieben, daher muss ich es studieren, aber zuerst werde ich einen Wrapper in Python schreiben, die Spezifikationen des Standardtreibers studieren und den Implementierungsablauf zusammenfassen.
--python Treiber für MySQL --python db API-Spezifikationen
Das Folgende ist eine Zusammenfassung dessen, was ich versucht habe zu implementieren, während ich diesmal naiv war.
Code-weise mit unserem Treiber
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()
Ich möchte so etwas tun.
Hier ist `` `MySQLdbdie Bibliothek der Treiber, die wir bereitstellen,mysqldbAberconnection```Erstellen Sie ein Objekt der Klasse mit den erforderlichen Konfigurationsvariablen.
Außerdem erstellt die Klasse "Verbindung" ein Objekt der Klasse "Cursor".
Die Cursor-Klasse ist Cursor.execute()Oder Cursor.fetchone()Hat eine Methode wie
Die Ergebnisse der übergebenen SQL werden gespeichert und Sie können auf diese Ergebnisse zugreifen und sie anzeigen.
Wenn Einfügen oder Aktualisieren ausgeführt wird,
#### **`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()
Auf diese Weise wird die geänderte Operation tatsächlich in der Datenbank wiedergegeben, indem die Methode `commit ()` der db.coimmit () , dh der connection```-Klasse, aufgerufen wird. Wenn Sie es zurückbekommen möchten, können Sie es mit einer Methode wie `db.rollback ()` tun.
Außerdem wird die Klasse `Error``` implementiert, um Fehler auf eine bestimmte Weise zu behandeln. Darüber hinaus wird die Klasse `Type``` verwendet, um den zurückgegebenen Datentyp genau in den entsprechenden Datentyp von Python zu ändern, z. B. wenn das Backend der DB-Operation nicht in Python geschrieben ist. Ich werde.
Auf der Grundlage des oben Gesagten werde ich kurz vorstellen, was ich implementiert habe.
Im Hauptverzeichnis
tree .
Das Ergebnis von
.
├── api_core
│ ├── api_connection.py
│ ├── api_cursor.py
│ ├── api_error.py
│ ├── api_type.py
│ ├── __init__.py
│ └── our_api.py
├── main.py
└── requirements.txt
Es ist aus.
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)
Ich freue mich darauf zu sehen, wie dies aussieht, wenn es in Java geschrieben ist.
Ende.
Recommended Posts