首页 > 代码库 > python -- peewee处理数据库连接

python -- peewee处理数据库连接

目前,实现了的Database子类有三个;SqliteDatabaseMySQLDatabasePostgresqlDatabase

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 = http://www.mamicode.com/excute_sql(TB1, self.select1)
    for re in data:
print re.id  

 运行test_db.py,可以打印出数据库查询结果,其它数据库操作均可使用该方法

python -- peewee处理数据库连接