我的第一個python web開發框架（11）——工具函數包說明（二）

　　db_helper.py是數據庫操作包，主要有兩個函數，分別是read()數據庫讀操作函數和write()數據庫寫操作函數。這個包的代碼是從小戴同學分享的博文改造過來的。

#!/usr/bin/env python
# coding=utf-8

import psycopg2
from common import log_helper
from config import const

# 初始化數據庫參數
db_name = const.DB_NAME
db_host = const.DB_HOST
db_port = const.DB_PORT
db_user = const.DB_USER
db_pass = const.DB_PASS


def read(sql):
    """
    連接pg數據庫並進行數據查詢
    如果連接失敗，會把錯誤寫入日志中，並返回false，如果sql執行失敗，也會把錯誤寫入日志中，並返回false
    如果所有執行正常，則返回查詢到的數據，這個數據是經過轉換的，轉成字典格式，方便模板調用，其中字典的key是數據表里的字段名
    """
    try:
        # 連接數據庫
        conn = psycopg2.connect(database=db_name, user=db_user, password=db_pass, host=db_host, port=db_port)
        # 獲取游標
        cursor = conn.cursor()
    except Exception as e:
        print(e.args)
        log_helper.error('連接數據庫失敗：' + str(e.args))
        return False
    try:
        # 執行查詢操作
        cursor.execute(sql)
        # 將返回的結果轉換成字典格式
        data = [dict((cursor.description[i][0], value) for i, value in enumerate(row)) for row in cursor.fetchall()]
    except Exception as e:
        print(e.args)
        log_helper.error('sql執行失敗:' + str(e.args) + ' sql:' + str(sql))
        return False
    finally:
        # 關閉游標和數據庫鏈接
        cursor.close()
        conn.close()
    # 返回結果（字典格式）
    return data


def write(sql, vars):
    """
    連接pg數據庫並進行寫的操作
    如果連接失敗，會把錯誤寫入日志中，並返回false，如果sql執行失敗，也會把錯誤寫入日志中，並返回false，如果所有執行正常，則返回true
    """
    try:
        # 連接數據庫
        conn = psycopg2.connect(database=db_name, user=db_user, password=db_pass, host=db_host, port=db_port)
        # 獲取游標
        cursor = conn.cursor()
    except Exception as e:
        print(e.args)
        log_helper.error('連接數據庫失敗：' + str(e.args))
        return False
    try:
        # 執行sql語句
        cursor.execute(sql, vars)
        # 提交事務
        conn.commit()
    except Exception as e:
        print(e.args)
        # 如果出錯，則事務回滾
        conn.rollback()
        log_helper.error('sql執行失敗:' + str(e.args) + ' sql:' + str(sql))
        return False
    else:
        # 獲取數據
        try:
            data = [dict((cursor.description[i][0], value) for i, value in enumerate(row))
                         for row in cursor.fetchall()]
        except Exception as e:
            # 沒有設置returning或執行修改或刪除語句時，記錄不存在
            data = None
    finally:
        # 關閉游標和數據庫鏈接
        cursor.close()
        conn.close()

    # 如果寫入數據后，將數據庫返回的數據返回給調用者
    return data

　　read(sql)是用來執行數據庫查詢操作，里面沒有事務提交，所以用它來執行增刪改操作時，雖然能提交成功，但執行后數據庫記錄也不會有什么變化，所以只能用它來執行select語句

　　write(sql, data)是用來執行數據庫寫操作的，write函數執行后會返回下面幾種狀態：

　　1.False狀態（數據庫鏈接失敗、sql語句不正確、鏈接數據庫操時等執行數據庫出現異常時返回這個狀態）

　　2.None狀態（sql語句沒有添加RETURNING id代碼指定sql語句執行結束后返回指定字段值時出現；你如果修改代碼第80行，data = None為data = True，執行成功時則會返回True狀態）

　　3.[] （sql語句添加了returning函數，且執行修改或刪除時，記錄不存在）

　　4.{'id': 1,}（sql語句添加了returning函數，執行成功后返回我們指定的字段值）

　　PS：我們在執行新增的時候，如果想要獲取新增的id，postgresql有一個非常好用的函數returning，只需要在增刪改語句的后面添加returning id或returning id,name或returning *等你想要返回的字段名稱，語句執行成功以后都會返回這些指定的字段值，大家可以嘗試修改測試用例代碼，看看返回的值是什么。

#!/usr/bin/evn python
# coding=utf-8

import unittest
from common import db_helper


class DbHelperTest(unittest.TestCase):
    """數據庫操作包測試類"""

    def setUp(self):
        """初始化測試環境"""
        print('------ini------')

    def tearDown(self):
        """清理測試環境"""
        print('------clear------')

    def test(self):
        # 新增記錄，不帶return參數
        sql = """
            INSERT INTO product_class(
              name, is_enable)
            VALUES (%s, %s)
        """
        data = ('糖果', 1)
        result = db_helper.write(sql, data)
        print(result)

        # 新增記錄，使用return參數返回新增id
        sql = """
            INSERT INTO product_class(
              name, is_enable)
            VALUES (%s, %s)
            RETURNING id;
        """
        data = ('餅干', 1)
        result = db_helper.write(sql, data)
        print(result)

        # 修改不存在的記錄
        sql = """
            UPDATE product_class
               SET name=%s, is_enable=%s
            WHERE id=10000
            RETURNING id;
        """
        data = ('糖果', 1)
        result = db_helper.write(sql, data)
        print(result)

        # 查詢記錄
        sql = """
            SELECT * FROM product_class
        """
        result = db_helper.read(sql)
        print(result)

if __name__ == '__main__':
    unittest.main()

　　執行結果

------ini------
None
[{'id': 2}]
[]
[{'id': 1, 'name': '糖果', 'add_time': datetime.datetime(2017, 10, 16, 14, 51, 49), 'is_enable': 1}, {'id': 2, 'name': '餅干', 'add_time': datetime.datetime(2017, 10, 16, 15, 30, 50), 'is_enable': 1}]
------clear------

　　

 

　　encrypt_helper.py是加密操作包，目前只有md5加密函數，其他加密函數以后有需要再添加進來

#!/usr/bin/evn python
# coding=utf-8

import hashlib

def md5(text):
    """md5加密函數"""
    md5 = hashlib.md5()
    if not isinstance(text, bytes):
        text = str(text).encode('utf-8')
    md5.update(text)
    return md5.hexdigest()

　　md5()參數類型支持各種類型，如果參數為非bytes類型時會自動轉換為str類型來進行操作，例如以下測試用例，大家也可以嘗試用元組、字典或列表類型測試看看結果

#!/usr/bin/evn python
# coding=utf-8

import unittest
from common import encrypt_helper


class DbHelperTest(unittest.TestCase):
    """數據庫操作包測試類"""

    def setUp(self):
        """初始化測試環境"""
        print('------ini------')

    def tearDown(self):
        """清理測試環境"""
        print('------clear------')

    def test(self):
        result = encrypt_helper.md5(1)
        print(result)
        self.assertEqual(result, 'c4ca4238a0b923820dcc509a6f75849b')

        result = encrypt_helper.md5('1')
        print(result)
        self.assertEqual(result, 'c4ca4238a0b923820dcc509a6f75849b')

        result = encrypt_helper.md5(b'1')
        print(result)
        self.assertEqual(result, 'c4ca4238a0b923820dcc509a6f75849b')

if __name__ == '__main__':
    unittest.main()

　　執行結果

------ini------
c4ca4238a0b923820dcc509a6f75849b
c4ca4238a0b923820dcc509a6f75849b
c4ca4238a0b923820dcc509a6f75849b
------clear------

 

 

　　except_helper.py包主要功能是獲取代碼當前位置的堆棧信息，它被log_helper.py包的error()錯誤日志記錄函數調用，輸出發生錯誤時的堆棧信息內容，方便開發人員分析代碼異常。

#!/usr/bin/evn python
# coding=utf-8

import os
import sys

def detailtrace():
    """獲取程序當前運行的堆棧信息"""
    retStr = ""
    f = sys._getframe()
    f = f.f_back        # first frame is detailtrace, ignore it
    while hasattr(f, "f_code"):
        co = f.f_code
        retStr = "%s(%s:%s)->"%(os.path.basename(co.co_filename),
                  co.co_name,
                  f.f_lineno) + retStr
        f = f.f_back
    return retStr

 

 

　　json_helper.py包里只有一個日期格式化類。

#!/usr/bin/evn python
# coding=utf-8

import json
import datetime

class CJsonEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime.datetime):
            return obj.strftime('%Y-%m-%d %H:%M:%S')
        elif isinstance(obj, datetime.date):
            return obj.strftime('%Y-%m-%d')
        else:
            return json.JSONEncoder.default(self, obj)

 　　python的json將時間類型轉換為字符串時，它會處理不了出現異常，需要使用這個自定義類進行格式化處理

　　比如我們如果直接這樣對時間類型進行轉換時，就會出現異常：

    def test(self):
        js = {
            'test5': datetime.datetime.now(),
        }
        print(js)
        result = json.dumps(js)
        print(result)

　　執行結果：

------ini------
{'test5': datetime.datetime(2017, 10, 16, 16, 56, 58, 654832)}
------clear------

Error
Traceback (most recent call last):
  File "E:\Python\simple\code\test\json_helper_test.py", line 26, in test
    result = json.dumps(js)
  File "C:\Users\Empty\AppData\Local\Programs\Python\Python35-32\lib\json\__init__.py", line 230, in dumps
    return _default_encoder.encode(obj)
  File "C:\Users\Empty\AppData\Local\Programs\Python\Python35-32\lib\json\encoder.py", line 198, in encode
    chunks = self.iterencode(o, _one_shot=True)
  File "C:\Users\Empty\AppData\Local\Programs\Python\Python35-32\lib\json\encoder.py", line 256, in iterencode
    return _iterencode(o, 0)
  File "C:\Users\Empty\AppData\Local\Programs\Python\Python35-32\lib\json\encoder.py", line 179, in default
    raise TypeError(repr(o) + " is not JSON serializable")
TypeError: datetime.datetime(2017, 10, 16, 16, 56, 58, 654832) is not JSON serializable

　　改成下面代碼的話就正常了

    def test(self):
        js = {
            'test5': datetime.datetime.now(),
        }
        print(js)
        result = json.dumps(js, cls=json_helper.CJsonEncoder)
        print(result)

　　執行結果

------ini------
{'test5': datetime.datetime(2017, 10, 16, 16, 59, 40, 756103)}
{"test5": "2017-10-16 16:59:40"}
------clear------

 

 

　　本文對應的源碼下載

 

版權聲明：本文原創發表於 博客園，作者為 AllEmpty 本文歡迎轉載，但未經作者同意必須保留此段聲明，且在文章頁面明顯位置給出原文連接，否則視為侵權。

python開發QQ群：669058475（本群已滿）、733466321（可以加2群）    作者博客：http://www.cnblogs.com/EmptyFS/