APIExing/dbimpl.py at master · baolingfeng/APIExing · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
#!/usr/bin/env
# -*- coding: utf8 -*-

# import requests
# import json
# from requests.auth import HTTPBasicAuth
import sqlite3
# import logging
import mysql.connector


class DBImpl:

    def __init__(self, config):
        self.type = config['type'] if 'type' in config else 'sqlite'
        self.url = config['url']
        self.username = config['username'] if 'username' in config else None
        self.password = config['password'] if 'password' in config else None
        self.port = config['port'] if 'port' in config else 3306
        self.database = config['database'] if 'database' in config else None

        self.conn = self.connection()

    def connection(self):
        if self.type == 'sqlite' or self.type is None:
            return sqlite3.connect(self.url)
        elif self.type == 'mysql':
            return mysql.connector.connect(user=self.username, password=self.password,
                                           host=self.url,
                                           port=self.port,
                                           database=self.database)
        else:
            raise Exception('the database is not supported')

    def close(self):
        self.conn.close()

    def create_table(self, table_name, sql, drop=True):
        # with self.conn:
        cur = self.conn.cursor()

        if drop:
            cur.execute("DROP TABLE IF EXISTS %s" % table_name)

        cur.execute(sql)

        cur.close()
        self.conn.commit()

    def create_table2(self, table_name, columns, drop=True):
        strsql = 'CREATE TABLE %s' % table_name
        strsql += '(' + ','.join([c['name'] + ' ' + c['type']
                                  for c in columns]) + ')'

        self.create_table(table_name, strsql, drop)

    def table_exist(self, table_name):
        # with self.conn:
        cur = self.conn.cursor()

        if self.type == 'sqlite':
            cur.execute(
                'SELECT name FROM sqlite_master WHERE type=\'table\' AND name = ?', (table_name,))
        elif self.type == 'mysql':
            cur.execute('SHOW TABLES LIKE %s', (table_name,))

        data = cur.fetchone()
        cur.close()
        return data is not None

    def queryone(self, sql, *args):
        # with self.conn:
        cur = self.conn.cursor()

        cur.execute(sql, args)

        data = cur.fetchone()

        cur.close()
        # conn.commit()
        return data

    def querymany(self, sql, *args):
        # with self.conn:
        cur = self.conn.cursor()

        cur.execute(sql, args)

        data = cur.fetchall()

        cur.close()
        # conn.commit()
        return data

    def updateone(self, sql, *args):
        # with self.conn:
        cur = self.conn.cursor()
        cur.execute(sql, args)

        cur.close()
        self.conn.commit()

    def updatemany(self, sql, objs):
        # with self.conn:
        cur = self.conn.cursor()
        cur.executemany(sql, objs)

        cur.close()
        self.conn.commit()


if __name__ == '__main__':
    # print(sqlite3)
    db = DBImpl({"url": "127.0.0.1", "username": "root",
                 "password": "123456", "database": "link_api"})

    print db.querymany('SELECT * FROM link_api.link_api_record where name = ?', 'aggregate')