python -- peewee處理數據庫連接

目前，實現了的Database子類有三個；SqliteDatabase、MySQLDatabase、PostgresqlDatabase

class  SqliteDatabase ( Database )

 “sqlite3”模塊驅動

class  MySQLDatabase ( Database )

“MySQLdb” 模塊驅動

class  PostgresqlDatabase ( Database )

“psycopg2” 模塊驅動

 

例如mysql數據庫連接,新建文件mysql_db.py：(其它兩種數據庫連接類似)

from peewee import *

database = MySQLDatabase(database, **kwargs)

database.connect()
print database.get_conn()

class BaseModel(Model):
    class Meta:
        database = database

class TB1(BaseModel):
    id = BigIntegerField(primary_key=True)

    class Meta:
        db_table = 'tb1'

其中database為mysql的Default Schema（默認數據庫），kwargs為連接數據庫的信息

格式如下：

{'host': 'vpca-1.vm.elenet.me', 'password': '123456', 'port': 9707, 'user': 'vpca_user'}

ps: 可以運行mysql_db.py，測試數據庫是否連接成功，結果如下：

　　<pymysql.connections.Connection object at 0x104c5d710>

 

然后新建一個操作數據庫的python文件 db_utils.py

def select_sql(table, sql):
    query_result = table.get(sql)
    return query_result

def select_all(table, sql):
    query_result = table.select().where(sql)
    return query_result

def excute_sql(table, sql):
    query_result = table.raw(sql)
    return query_result.execute()

def raw_sql(table, sql):
    query = table.raw(sql)
    return query

接下來，我們就可以對數據庫進行操作了，新建一個test_db.py文件,如下所示：

import unittest
from db_utils import excute_sql
from mysql_db import TB1

class TestDB(unittest.TestCase):
    def setUp(self):
　　　　self.select1 = "SELECT id FROM tb1 LIMIT 1"

　　def test_select_sql(self):
　　　　data = excute_sql(TB1, self.select1)
　　　　for re in data:
          print re.id  

 運行test_db.py，可以打印出數據庫查詢結果，其它數據庫操作均可使用該方法