Python 数据库适配器示例_python配置双数据源

Python 数据库适配器示例

代码

# -*- coding: utf-8 -*-
# @Time    : 2019/6/20 16:48
# @Author  : MLee
# @File    : ushuffle_dbU.py

import os
from distutils.log import warn as printf
from random import randrange as rand

if isinstance(__builtins__, dict) and "raw_input" in __builtins__:
    scanf = raw_input
elif hasattr(__builtins__, "raw_input"):
    scanf = raw_input
else:
    scanf = input

COLSIZ = 10
FILEDS = ("login", "userid", "projid")
RDBMSs = {"s": "sqlite", "m": "mysql", "g": "gadfly"}
DBNAME = "test"
DBUSER = "root"
DB_EXC = None
NAMELEN = 16

tformat = lambda s: str(s).title().ljust(COLSIZ)
cformat = lambda s: s.upper().ljust(COLSIZ)


def setup():
    return RDBMSs[scanf("""
Choose a database system:

(M)ySQL
(G)adfly
(S)QLite

Enter choice: """).strip().lower()[0]]


def connect(db, DBNAME=DBNAME):
    global DB_EXC
    dbDir = "%s_%s" % (db, DBNAME)
    if db == "sqlite":
        try:
            import sqlite3
        except ImportError:
            try:
                from pysqlite2 import dbapit2 as sqlite3
            except ImportError:
                return None

        DB_EXC = sqlite3
        if not os.path.isdir(dbDir):
            os.mkdir(dbDir)
        cxn = sqlite3.connect(os.path.join(dbDir, DBNAME))

    elif db == "mysql":
        try:
            import MySQLdb
            import _mysql_excptions as DB_EXC

            try:
                cxn = MySQLdb.connect(db=DBNAME)
            except DB_EXC.OperationalError:
                try:
                    cxn = MySQLdb.connect(user=DBUSER)
                    cxn.query('CREATE DATABASE %s' % DBNAME)
                    cxn.commit()
                    cxn.close()
                    cxn = MySQLdb.connect(db=DBNAME)
                except DB_EXC.OperationalError:
                    return None
        except ImportError:
            return None

    elif db == "gadfly":
        try:
            from gadfly import gadfly
            DB_EXC = gadfly
        except ImportError:
            return None

        try:
            cxn = gadfly(DBNAME, dbDir)
        except IOError:
            cxn = gadfly()
            if not os.path.isdir(dbDir):
                os.mkdir(DBNAME, dbDir)

    else:
        return None
    return cxn


def create(cur):
    try:
        cur.execute("""CREATE TABLE users (
          login VARCHAR(%d),
          userid    INTEGER,
          projid    INTEGER)
        """ % NAMELEN)
    except DB_EXC.OperationalError as e:
        drop(cur)
        create(cur)


drop = lambda cur: cur.execute("DROP TABLE users")

NAMES = (
    ('aaron', 8312), ('angela', 7603), ('deva', 7306),
    ('davina', 7902), ('elliot', 7911), ('ernie', 7410),
    ('jess', 7912), ('jim', 7512), ('larry', 7311),
)


def randName():
    pick = set(NAMES)
    while pick:
        yield pick.pop()


def insert(cur, db):
    if db == 'sqlite':
        cur.executemany("INSERT INTO users VALUES (?,?,?)",
                        [(who, uid, rand(1, 5)) for who, uid in randName()])
    elif db == "gadfly":
        for who, uid in randName():
            cur.execute("INSERT INTO users VALUES (?,?,？)",
                        (who, uid, rand(1, 5)))
    elif db == 'mysql':
        cur.executemany("INSERT INTO users VALUES (%s,%s,%s)",
                        [(who, uid, rand(1, 5)) for who, uid in randName()])


getRC = lambda cur: cur.rowcount if hasattr(cur, "rowcount") else -1


def update(cur):
    fr = rand(1, 5)
    to = rand(1, 5)
    cur.execute("UPDATE users SET projid=%d WHERE projid=%d" % (to, fr))
    return fr, to, getRC(cur)


def delete(cur):
    rm = rand(1, 5)
    cur.execute("DELETE FROM users WHERE projid=%d" % rm)
    return rm, getRC(cur)


def dbDump(cur):
    cur.execute("SELECT * FROM users")
    printf("\n%s" % ''.join(map(cformat, FILEDS)))
    for data in cur.fetchall():
        printf("".join(map(tformat, data)))


def main():
    db = setup()
    printf("*** Connect to %r database" % db)
    cxn = connect(db)
    if not cxn:
        printf("ERROR: %r not supported or unreachable, exit" % db)
        return

    cur = cxn.cursor()

    printf("\n*** Creating users table")
    create(cur)

    printf("\n*** Inserting names into table")
    insert(cur, db)
    dbDump(cur)

    printf("\n*** Randomly moving folks")
    fr, to, num = update(cur)
    printf("\t(%d users moved) from (%d) to (%d)" % (num, fr, to))
    dbDump(cur)

    printf("\n*** Randomly choosing group")
    rm, num = delete(cur)
    printf("\t(group #%d; %d users removed)" % (rm, num))
    dbDump(cur)

    printf("\n*** Dropping users table")
    drop(cur)
    printf("\n*** Close cxns")
    cur.close()
    cxn.commit()
    cxn.close()


if __name__ == '__main__':
    main()
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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194

# 运行结果：
Choose a database system:

(M)ySQL
(G)adfly
(S)QLite

Enter choice: s
*** Connect to 'sqlite' database

*** Creating users table

*** Inserting names into table

LOGIN     USERID    PROJID
Deva      7306      4
Elliot    7911      2
Ernie     7410      2
Jim       7512      2
Davina    7902      1
Aaron     8312      4
Larry     7311      1
Angela    7603      1
Jess      7912      2

*** Randomly moving folks
	(2 users moved) from (4) to (3)

LOGIN     USERID    PROJID
Deva      7306      3
Elliot    7911      2
Ernie     7410      2
Jim       7512      2
Davina    7902      1
Aaron     8312      3
Larry     7311      1
Angela    7603      1
Jess      7912      2

*** Randomly choosing group
	(group #2; 4 users removed)

LOGIN     USERID    PROJID
Deva      7306      3
Davina    7902      1
Aaron     8312      3
Larry     7311      1
Angela    7603      1

*** Dropping users table

*** Close cxns
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

啰嗦两句

这段代码摘自《Python 核心编程（第3版）》的 220 页。为什么要贴这个代码呢？不是因为其实现的数据库适配器的功能，而是在看这段代码时，发现有自己平时没怎么用过的编程技巧。想记录下来，方便以后查阅。

代码中的一些技巧

1. 关于输入输出，这里将输入封装为类似 C/C++ 的 scanf，同时，同时为了兼容 Python 2 和 Python 3，做了导入检查
2. 数据库适配器的导入做了兼容 Python 2 和 Python 3 的处理。
3. drop 函数直接用 lambda 表达式实现
4. randName 函数，破坏性的得到 NAMES 数据结构的中的值，采用了先将值复制到另一个数据结构，然后用生成器修改并返回值