python -- peewee处理数据库连接

Posted 2020-08-23

目前，实现了的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，可以打印出数据库查询结果，其它数据库操作均可使用该方法