參悟python元類(又稱metaclass)系列實戰(四)

寫在前面

在上一章節參悟python元類(又稱metaclass)系列實戰(三)完成了users類和users表的欄位映射;
在繼續豐富users類的操作之前, 我們花一章來完成mysql的鏈接;
有誤的地方懇請大神指正下，

創建全域連接池, 以求避免頻繁地打開和關閉資料庫連接

• 因為查詢耗時, 計劃引入異步, 即async/await(不懂沒關系, 就當是普通代碼, 只是呼叫方式特別而已)

  $ pip3 install aiomysql    # 通過pip安裝
  

• 創建Mysql類, 定義靜態方法createPool

  import aiomysql
  
  class Mysql:
      @staticmethod
      async def createPool():
          '''連接池由全域變數__pool存盤，預設情況下將編碼設定為utf8，自動提交事務'''
          try:
              global __pool
              __pool = await aiomysql.create_pool(
                  host='localhost',        # mysql服務器地址
                  port=3306,               # mysql監聽埠
                  user='root',             # mysql用戶名
                  password='passwd',       # mysql密碼
                  db='ormdemo',            # mysql實體名(database)
                  charset='utf8',          # 編碼設定
                  autocommit=True,         # 自動提交insert/update/delete
                  maxsize=10,              # 連接池上限
                  minsize=1                # 連接池下限
              )
              print('創建成功')
          except Exception as e:
              print(f'連接 mysql 出錯:{e}')
  
  # 測驗代碼
  if __name__ == '__main__':
      import asyncio
      loop = asyncio.get_event_loop()
      loop.run_until_complete(Mysql.createPool())
  

• 給Mysql類新增靜態方法select、execute

  @staticmethod
  async def select(sql, args, size=None):
      '''傳入SQL陳述句和SQL引數
      @sql :  sql陳述句
      @args:  條件值串列
      @size:  查詢的結果集大小
      '''
      print(f'{sql} ==> {args}')
      # 使用with背景關系管理器, 自動執行close
      async with __pool.acquire() as conn:
          # acquire: 從空閑池獲得連接，如果需要，并且池的大小小于maxsize，則創建新連接，
          async with conn.cursor(aiomysql.DictCursor) as cur:
              # cursor: 連接的游標
              # DictCursor: 作為字典回傳結果的游標
              await cur.execute(sql.replace('?', '%s'), args or ())
              # sql.replace('?', '%s'): 'SELECT * FROM users WHERE uid=?'.replace('?', '%s')
              # args or (): 'SELECT * FROM users WHERE uid=%s' % (101,)
              if size:
                  rs = await cur.fetchmany(size)
              else:
                  rs = await cur.fetchall()
          print(f'rows returned: {len(rs)}')
          return rs
  
  
  @staticmethod
  async def execute(sql, args, autocommit=True):
      '''
      INSERT、UPDATE、DELETE操作，可以通用的execute函式，
      因為這3種SQL的執行都需要相同的引數, 回傳一個整數表示影響的行數
      '''
      print(f'{sql} ==> {args}')
      async with __pool.acquire() as conn:
          if not autocommit:
              await conn.begin()           # 開始處理
          try:
              async with conn.cursor(aiomysql.DictCursor) as cur:
                  await cur.execute(sql.replace('?', '%s'), args)
                  affected = cur.rowcount  # 受影響的行數
              if not autocommit:
                  await conn.commit()      # 提交更改, 非自動提交時才需要
          except Exception as e:
              if not autocommit:
                  await conn.rollback()    # 回滾更改
              raise Exception(e)
          return affected
  

測驗下select

if __name__ == '__main__':
    import asyncio
    loop = asyncio.get_event_loop()
    loop.run_until_complete(Mysql.createPool())
    rs = loop.run_until_complete(Mysql.select('select uid, name from users where uid=?', [101]))
    print(rs)

• 控制臺輸出

創建成功
select * from users where uid=? ==> [101]
rows returned: 1
[{'uid': 101, 'name': 'z417'}]

總結

1. 定義了Mysql類, 其中有3個靜態方法

2. 引入了異步, 呼叫方式共分兩步

標籤：Python

上一篇：Python網路編程 —— 驗證客戶端的合法性

下一篇：python實作HmacSHA256加密演算法