Что такое хороший подход к управлению соединением db в приложении Python Google Cloud SQL (GAE)?

Question

Я изучаю Google App Engine и пытаюсь найти подходящий подход к управлению подключением базы данных к экземпляру Google Cloud SQL (если вы не использовали GC-SQL, в основном это MySQL в облаке, несколько ограничений).

Я использую среду GAE python (2.7) с инфраструктурой webapp2 для обработки запросов. Я знаю, что в FAQ говорится, что с каждым запросом рекомендуется создать новое соединение с БД, но я не знаю, какой рекомендуемый способ закрыть соединение. Каждый раз, когда я пытаюсь удалить таблицы во время разработки, GC-SQL зависает, и «show processlist» показывает, что существует множество процессов (вероятно, из-за того, что я не закрываю DB) и что один из них ждет блокировки (вероятно, процесс, пытающийся отбросить таблицы). Это раздражает и заставляет меня перезапустить экземпляр GC-SQL (например, перезапустить службу mysql-сервера, я полагаю). Есть также случайные икоты DB, которые, я считаю, связаны с тем фактом, что я действительно не закрываю свое соединение с БД.

Так, например, должен ли я иметь деструктор моего экземпляра подкаталога webapp2.Requesthandler для отключения от БД? Объекты GAE, похоже, иногда кэшируются, так что это тоже нужно рассмотреть. Полагаю, я мог бы просто подключиться/запросить/отключиться для каждого запроса, но это кажется субоптимальным.

Я знаю, что это неопределенный вопрос, но я надеюсь, что кто-то, кто играл в этой области, может пробраться несколько советов по моему пути.

Заранее благодарен!

Обновление: Я попытался реализовать оболочку вокруг методов, которым нужен курсор, используя ответ Шая в качестве отправной точки. Я получаю ошибки GAE. Вот новый вопрос, связанный с этим: What are the connection limits for Google Cloud SQL from App Engine, and how to best reuse DB connections?

Answer 1

Я не знаком с Google Cloud SQL, но разве вы не могли использовать связующее ПО WSGI для открытия и закрытия соединения?

Answer 2

Я написал декоратора для обработки соединения SQL, не стесняйтесь пламя :)

# Here is how you use the decorator from below 
# the open, commit, and close is done by the decorator 
@need_cursor(do_commit = True) 
def get(self, cursor, request): # cursor param is added by the decorator 
    execute_sql(cursor, sql) 

---

def need_cursor(do_commit = False): 
    def method_wrap(method): 
     def wrap(*args, **kwargs): 
      conn = os.environ.get("__data_conn") 

      # Recycling connection for the current request 
      # For some reason threading.local() didn't worked 
      # and yes os.environ suppose to be thread safe 
      if not conn:     
       conn = create_connection() # You need to implement this 
       os.environ["__data_conn"] = conn 
       os.environ["__data_conn_ref"] = 1 
      else: 
       os.environ["__data_conn_ref"] = 
        os.environ["__data_conn_ref"] + 1 

      try: 
       cursor = conn.cursor() 
       try: 
        result = method(cursor, *args, **kwargs) 

        if do_commit or os.environ.get("__data_conn_commit"): 
         os.environ["__data_conn_commit"] = False 
         conn.commit() 

        return result      
       finally: 
        cursor.close()     
      finally: 
       os.environ["__data_conn_ref"] = 
        os.environ["__data_conn_ref"] - 1 
       if os.environ["__data_conn_ref"] == 0: 
        os.environ["__data_conn"] = None 
        conn.close()   

     return wrap 

    return method_wrap 

Answer 3

Вот полный пример пример HelloWorld приложение из Getting Started Guide. Он основан на фрагментах от Shay Erlichmen и JJC, но эта версия является потокобезопасной.

Вы можете использовать его как это:

@with_db_cursor(do_commit = True) 
    def get(self, cursor): 
     cursor.execute('SELECT guestName, content, entryID FROM entries') 

---

app.yaml

application: helloworld 
version: 1 
runtime: python27 
api_version: 1 
threadsafe: true 

handlers: 
- url: /.* 
    script: helloworld.app 

---

helloworld.py

import cgi 
import logging 
import os 
import threading 
import webapp2 

from google.appengine.api import rdbms 

_INSTANCE_NAME = <name goes here> 

def _db_connect(): 
    return rdbms.connect(instance=_INSTANCE_NAME, database='guestbook') 

_mydata = threading.local() 

def with_db_cursor(do_commit = False): 
    """ Decorator for managing DB connection by wrapping around web calls. 

    Stores connections and open cursor count in a threadlocal 
    between calls. Sets a cursor variable in the wrapped function. Optionally 
    does a commit. Closes the cursor when wrapped method returns, and closes 
    the DB connection if there are no outstanding cursors. 

    If the wrapped method has a keyword argument 'existing_cursor', whose value 
    is non-False, this wrapper is bypassed, as it is assumed another cursor is 
    already in force because of an alternate call stack. 
    """ 
    def method_wrap(method): 
    def wrap(self, *args, **kwargs): 
     if kwargs.get('existing_cursor', False): 
     # Bypass everything if method called with existing open cursor. 
     return method(self, None, *args, **kwargs) 

     if not hasattr(_mydata, 'conn') or not _mydata.conn: 
     _mydata.conn = _db_connect() 
     _mydata.ref = 0 
     _mydata.commit = False 

     conn = _mydata.conn 
     _mydata.ref = _mydata.ref + 1 

     try: 
     cursor = conn.cursor() 
     try: 
      result = method(self, cursor, *args, **kwargs) 
      if do_commit or _mydata.commit: 
      _mydata.commit = False 
      conn.commit() 
      return result 
     finally: 
      cursor.close() 
     finally: 
     _mydata.ref = _mydata.ref - 1 
     if _mydata.ref == 0: 
      _mydata.conn = None 
      logging.info('Closing conn') 
      conn.close() 
    return wrap 
    return method_wrap 


class MainPage(webapp2.RequestHandler): 
    @with_db_cursor(do_commit = True) 
    def get(self, cursor): 
     cursor.execute('SELECT guestName, content, entryID FROM entries') 
     self.response.out.write(""" 
      <!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Strict//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-strict.dtd"> 
      <html xmlns="http://www.w3.org/1999/xhtml" lang="en" xml:lang="en"> 
      <head> 
       <title>My Guestbook!</title> 
      </head> 
      <body>""") 
     self.response.out.write(""" 
       <table style="border: 1px solid black"> 
       <tbody> 
        <tr> 
        <th width="35%" style="background-color: #CCFFCC; margin: 5px">Name</th> 
        <th style="background-color: #CCFFCC; margin: 5px">Message</th> 
        <th style="background-color: #CCFFCC; margin: 5px">ID</th> 
        </tr>""") 
     for row in cursor.fetchall(): 
      self.response.out.write('<tr><td>') 
      self.response.out.write(cgi.escape(row[0])) 
      self.response.out.write('</td><td>') 
      self.response.out.write(cgi.escape(row[1])) 
      self.response.out.write('</td><td>') 
      self.response.out.write(row[2]) 
      self.response.out.write('</td></tr>') 

     self.response.out.write(""" 
      </tbody> 
      </table> 
       <br /> No more messages! 
       <br /><strong>Sign the guestbook!</strong> 
       <form action="/sign" method="post"> 
       <div>First Name: <input type="text" name="fname" style="border: 1px solid black"></div> 
       <div>Message: <br /><textarea name="content" rows="3" cols="60"></textarea></div> 
       <div><input type="submit" value="Sign Guestbook"></div> 
      </form> 
      </body> 
     </html>""") 

class Guestbook(webapp2.RequestHandler): 
    @with_db_cursor(do_commit = True) 
    def post(self, cursor): 
    fname = self.request.get('fname') 
    content = self.request.get('content') 
    # Note that the only format string supported is %s 
    cursor.execute('INSERT INTO entries (guestName, content) VALUES (%s, %s)', (fname, content)) 

    self.redirect("/") 

app = webapp2.WSGIApplication(
    [('/', MainPage), 
    ('/sign', Guestbook)], 
    debug=True) 

Answer 4

Это мой подход, который рассматривает возможные исключения.Я использую этот подход в рабочей среде и хорошо работает:

def _create_connection(schema): 

    if (os.getenv('SERVER_SOFTWARE') and 
     os.getenv('SERVER_SOFTWARE').startswith('Google App Engine/')): 
     socket = '/cloudsql/%s' % env.DB_INSTANCE_NAME 
     return MySQLdb.connect(unix_socket=socket, user=env.DB_APP_USER, 
           passwd=env.DB_APP_PASS, db=schema) 
    else: 
     return MySQLdb.connect(host='127.0.0.1', port=3306, 
           user=env.DB_APP_USER, passwd=env.DB_APP_PASS, 
           db=schema) 


def with_db(commit=False, schema=env.DB_SCHEMA_NAME): 

    def method_wrap(method): 
     @functools.wraps(method) 
     def wrap(self, *args, **kwds): 
      # If needed,a connection pool can be added here. 
      connection = _create_connection(schema) 

      try: 
       cur = connection.cursor() 
       self.cur = cur 
       self.conn = connection 

       result = method(self, *args, **kwds) 

       if commit: 
        connection.commit() 

      except OperationalError as e: 

       logging.error('Operational error.\r\nSQL exception: {},\r\n' 
           'Last Query: {}'.format(e, cur._last_executed)) 

       if commit and connection.open: 
        connection.rollback() 
       raise 

      except MySQLError as e: 

       try: 
        warns = self.conn.show_warnings() 
        error = self.conn.error() 
       except: 
        warns = "" 
        error = "" 

       logging.error('Try to rolling back transaction.\r\nSQL exception: {},\r\n' 
           'Last Query: {},\r\nConn warn: {},\r\nError: {}' 
           .format(e, cur._last_executed, warns, error)) 


       if commit and connection.open: 
        connection.rollback() 
       raise 

      except Exception as e: 
       logging.error('Try to rolling back transaction. Non SQL exception: {0}'.format(e)) 

       if commit and connection.open: 
        connection.rollback() 
       raise 

      finally: 
       connection.close() 

      return result 
     return wrap 
    return method_wrap 

Вы можете использовать его как это:

@with_db(commit=True) 
def update_user_phone(self, user, phone): 
    self.cur.execute(_SQL_UPDATE_USER_PHONE, (phone, user.id)) 

    # add or replace existing user to cache 
    user.phone = phone 
    self._update_user_cache(user)