peewee操作mysql_python -- peewee处理数据库连接

目前，实现了的Database子类有三个；SqliteDatabase、MySQLDatabase、PostgresqlDatabase

classSqliteDatabase(Database)“sqlite3”模块驱动classMySQLDatabase(Database)“MySQLdb” 模块驱动classPostgresqlDatabase(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，测试数据库是否连接成功，结果如下：

然后新建一个操作数据库的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，可以打印出数据库查询结果，其它数据库操作均可使用该方法