The story of making a standard driver for db with python.


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


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

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,

#### **``**


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.
         VALUES ('Mac', 'Mohan', 20, 'M', 2000)"""
   # Execute the SQL command
   # Commit your changes in the database
   # Rollback in case there is any error

# disconnect from server

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
│   ├──
│   ├──
│   ├──
│   ├──
│   ├──
│   └──
└── requirements.txt

It is made up of.

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):

    def cursor(self):
        return Cursor(self.proxy_conf_dict)

    def close(self):

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 =, 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])

    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:

            url = f'http://{self.conf_dict["api_host"]}:{self.conf_dict["api_port"]}/query'
            result =, 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:
                return self.result[self.index]
                raise Exception('cursor index is not appropriate for result')

    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


    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

class Error:
        Error class
            TODO implementation
    def __init__(self):

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:
                parse_string_to: params(target_string:str, target_type:str)
                                 return either python datetime, date, time object which parsed string to.
    def __init__(self):

    ## some data type such as date, datetime, time is returned as string, needs to be converted to corresponding python object.

    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')
        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   

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):

    def connect(conf_dict=None, conf_file=None):
        if conf_dict:
            return Connection(conf_dict)

        elif conf_file:
            return Connection(conf_file)

in conclusion

I'm looking forward to seeing what this looks like when written in Java.


Recommended Posts

The story of making a standard driver for db with python.
The story of making a module that skips mail with python
The story of making a university 100 yen breakfast LINE bot with Python
The story of making a question box bot with
The story of making Python an exe
The story of blackjack A processing (python)
The story of low learning costs for Python
Image processing? The story of starting Python for
The story of making a lie news generator
The story of making a mel icon generator
The story of making a sound camera with Touch Designer and ReSpeaker
A story about making 3D space recognition with Python
The story of making a music generation neural network
A story about making Hanon-like sheet music with Python
The story of making a web application that records extensive reading with Django
The story of making a tool to load an image with Python ⇒ save it as another name
[Introduction to Python] How to get the index of data with a for statement
The story of creating a VIP channel for in-house chatwork
The story of implementing the popular Facebook Messenger Bot with python
[For beginners] Summary of standard input in Python (with explanation)
A story about an amateur making a breakout with python (kivy) ②
Create a Twitter BOT with the GoogleAppEngine SDK for Python
The story of rubyist struggling with python :: Dict data with pycall
A story about an amateur making a breakout with python (kivy) ①
[Python] The first step to making a game with Pyxel
Turn an array of strings with a for statement (Python3)
The story of Python and the story of NaN
The story of writing a program
A memorandum of understanding for the Python package management tool ez_setup
A story stuck with the installation of the machine learning library JAX
Control the motor with a motor driver using python on Raspberry Pi 3!
[python, ruby] fetch the contents of a web page with selenium-webdriver
A story that struggled to handle the Python package of PocketSphinx
Create a color picker for the color wheel with Python + Qt (PySide)
Read the standard output of a subprocess line by line in Python
I made a lot of files for RDP connection with Python
The idea of feeding the config file with a python file instead of yaml
Create a compatibility judgment program with the random module of python.
Check the existence of the file with python
The story of making an immutable mold
The story of releasing a Python text check tool on GitHub x CircleCI for the first time
The story of manipulating python global variables
[python] [meta] Is the type of python a type?
UnicodeEncodeError struggle with standard output of python3
The third night of the loop with for
Pandas of the beginner, by the beginner, for the beginner [Python]
The second night of the loop with for
The story of creating a bot that displays active members in a specific channel of slack with python
A story that visualizes the present of Qiita with Qiita API + Elasticsearch + Kibana
The story of a Parking Sensor in 10 minutes with GrovePi + Starter Kit
[AtCoder explanation] Control the A, B, C problems of ABC182 with Python!
Calculate the shortest route of a graph with Dijkstra's algorithm and Python
[Introduction to Udemy Python3 + Application] 47. Process the dictionary with a for statement
Get the number of searches with a regular expression. SeleniumBasic VBA Python
The story of having a hard time introducing OpenCV with M1 MAC
[AtCoder explanation] Control the A, B, C problems of ABC186 with Python!
[AtCoder explanation] Control the A, B, C problems of ABC185 with Python!
Calculate the probability of being a squid coin with Bayes' theorem [python]
Hit a method of a class instance with the Python Bottle Web API
Receive a list of the results of parallel processing in Python with starmap
I tried to streamline the standard role of new employees with Python