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

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

 1 #!/usr/bin/env python
 2 # coding=utf-8
 3 
 4 import psycopg2
 5 from common import log_helper
 6 from config import const
 7 
 8 # 初始化數據庫參數
 9 db_name = const.DB_NAME
10 db_host = const.DB_HOST
11 db_port = const.DB_PORT
12 db_user = const.DB_USER
13 db_pass = const.DB_PASS
14 
15 
16 def read(sql):
17     """
18     鏈接pg數據庫並進行數據查詢
19     若是鏈接失敗，會把錯誤寫入日誌中，並返回false，若是sql執行失敗，也會把錯誤寫入日誌中，並返回false
20     若是全部執行正常，則返回查詢到的數據，這個數據是通過轉換的，轉成字典格式，方便模板調用，其中字典的key是數據表裏的字段名
21     """
22     try:
23         # 鏈接數據庫
24         conn = psycopg2.connect(database=db_name, user=db_user, password=db_pass, host=db_host, port=db_port)
25         # 獲取遊標
26         cursor = conn.cursor()
27     except Exception as e:
28         print(e.args)
29         log_helper.error('鏈接數據庫失敗：' + str(e.args))
30         return False
31     try:
32         # 執行查詢操做
33         cursor.execute(sql)
34         # 將返回的結果轉換成字典格式
35         data = [dict((cursor.description[i][0], value) for i, value in enumerate(row)) for row in cursor.fetchall()]
36     except Exception as e:
37         print(e.args)
38         log_helper.error('sql執行失敗:' + str(e.args) + ' sql:' + str(sql))
39         return False
40     finally:
41         # 關閉遊標和數據庫連接
42         cursor.close()
43         conn.close()
44     # 返回結果（字典格式）
45     return data
46 
47 
48 def write(sql, vars):
49     """
50     鏈接pg數據庫並進行寫的操做
51     若是鏈接失敗，會把錯誤寫入日誌中，並返回false，若是sql執行失敗，也會把錯誤寫入日誌中，並返回false，若是全部執行正常，則返回true
52     """
53     try:
54         # 鏈接數據庫
55         conn = psycopg2.connect(database=db_name, user=db_user, password=db_pass, host=db_host, port=db_port)
56         # 獲取遊標
57         cursor = conn.cursor()
58     except Exception as e:
59         print(e.args)
60         log_helper.error('鏈接數據庫失敗：' + str(e.args))
61         return False
62     try:
63         # 執行sql語句
64         cursor.execute(sql, vars)
65         # 提交事務
66         conn.commit()
67     except Exception as e:
68         print(e.args)
69         # 若是出錯，則事務回滾
70         conn.rollback()
71         log_helper.error('sql執行失敗:' + str(e.args) + ' sql:' + str(sql))
72         return False
73     else:
74         # 獲取數據
75         try:
76             data = [dict((cursor.description[i][0], value) for i, value in enumerate(row))
77                          for row in cursor.fetchall()]
78         except Exception as e:
79             # 沒有設置returning或執行修改或刪除語句時，記錄不存在
80             data = None
81     finally:
82         # 關閉遊標和數據庫連接
83         cursor.close()
84         conn.close()
85 
86     # 若是寫入數據後，將數據庫返回的數據返回給調用者
87     return data

View Code

　　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/