Understanding Python metaclass series (4)

Write it at the front

In the previous chapter Understanding python The metaclass ( also called metaclass) Series of actual combat ( 3、 ... and ) It's done users class and users surface Field mapping for ;
Continue to enrich users class Before the operation of , It took us a chapter to complete mysql Link to ;
Please point out the mistakes .

Create a global connection pool , To avoid frequently opening and closing database connections

• Because the query takes time , Plan to introduce asynchrony , namely async/await( I don't understand it doesn't matter , Think of it as plain code , It's just that it's called in a special way )

  $ pip3 install aiomysql # adopt pip install
  

• establish Mysql class , Define static methods createPool

  import aiomysql
  class Mysql:
  @staticmethod
  async def createPool():
  ''' The connection pool is made up of global variables __pool Storage , By default, the encoding is set to utf8, Auto commit transaction '''
  try:
  global __pool
  __pool = await aiomysql.create_pool(
  host='localhost', # mysql Server address
  port=3306, # mysql Listening port
  user='root', # mysql user name
  password='passwd', # mysql password
  db='ormdemo', # mysql Instance name (database)
  charset='utf8', # Encoding settings
  autocommit=True, # Automatic submission insert/update/delete
  maxsize=10, # Connection pool upper limit
  minsize=1 # Connection pool lower limit
  )
  print(' Create success ')
  except Exception as e:
  print(f' Connect mysql error :{e}')
  # Test code
  if __name__ == '__main__':
  import asyncio
  loop = asyncio.get_event_loop()
  loop.run_until_complete(Mysql.createPool())
  

• to Mysql class New static method select、execute

  @staticmethod
  async def select(sql, args, size=None):
  ''' Pass in SQL Statement and SQL Parameters
  @sql : sql sentence
  @args: List of condition values
  @size: The result set size of the query
  '''
  print(f'{sql} ==> {args}')
  # Use with Context manager , Automatic execution close
  async with __pool.acquire() as conn:
  # acquire: Get connections from the free pool . if necessary , And the size of the pool is smaller than maxsize, Create a new connection .
  async with conn.cursor(aiomysql.DictCursor) as cur:
  # cursor: Connected cursor
  # DictCursor: A cursor that returns the result as a dictionary
  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 operation , It's universal execute function ,
  Because of this 3 Kind of SQL The same parameters are required for the execution of , Returns an integer representing the number of rows affected
  '''
  print(f'{sql} ==> {args}')
  async with __pool.acquire() as conn:
  if not autocommit:
  await conn.begin() # Start to deal with
  try:
  async with conn.cursor(aiomysql.DictCursor) as cur:
  await cur.execute(sql.replace('?', '%s'), args)
  affected = cur.rowcount # Rows affected
  if not autocommit:
  await conn.commit() # Submit changes , Only when it is not submitted automatically
  except Exception as e:
  if not autocommit:
  await conn.rollback() # Roll back changes
  raise Exception(e)
  return affected
  

Under test 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)

• Console output

 Create success
select * from users where uid=? ==> [101]
rows returned: 1
[{'uid': 101, 'name': 'z417'}]

summary

1. Defined Mysql class , Among them is 3 A static method

2. Asynchronous is introduced , The calling method is divided into two steps