0

我正在 python 中实现一个数据库连接器类。我将使用retrytenacity 库中的装饰器在超时时重试数据库连接。

我想将self.retry_countand传递给装饰器self.retry_interval中的参数。retry

## etl_connect.py
from sqlalchemy import create_engine
import pymysql
import logging
from tenacity import *

class Connector():
    def __init__(self, mode, conn_str, retry_count, retry_interval):
        self.mode = mode
        self.conn_str = conn_str
        self.retry_count = retry_count
        self.retry_interval = retry_interval
        self.engine = None
        self.conn = None

    @retry(wait=wait_fixed(self.retry_interval), stop=stop_after_attempt(self.retry_count))
    def mysql_connect(self):
        logging.info('Connecting to mysql. (retry count=%d)' % (self.mysql_connect.retry.statistics['attempt_number']))
        mysql_engine = create_engine(self.conn_str)
        mysql_conn = mysql_engine.connect()
        logging.info('Connected to mysql successfully with %d attempt(s).' % (self.mysql_connect.retry.statistics['attempt_number']))
        return (mysql_engine, mysql_conn)

现在调用mysql_connect函数：

## call.py
from etl_connect import *
mysql_connector = Connector('mysql', 'mysql database string here', 5, 10)
engine, conn = mysql_connector.mysql_connect()

但它显示：NameError: name 'self' is not defined.

Traceback (most recent call last):
  File "call.py", line 5, in <module>
    from etl_connect import *
  File "/home/developer/ETL_modules/etl_connect.py", line 19, in <module>
    class Connector():
  File "/home/developer/ETL_modules/etl_connect.py", line 56, in Connector
    @retry(wait=wait_fixed(self.retry_interval), stop=stop_after_attempt(self.retry_count))
NameError: name 'self' is not defined

有什么方法可以将self.retry_count&传递self.retry_interval给装饰者吗？

4

4 回答 4

1

不是装饰方法，而是在调用方法retry时调用。

## etl_connect.py
from sqlalchemy import create_engine
import pymysql
import logging
from tenacity import *

class Connector():
    def __init__(self, mode, conn_str, retry_count, retry_interval):
        self.mode = mode
        self.conn_str = conn_str
        self.retry_count = retry_count
        self.retry_interval = retry_interval
        self.engine = None
        self.conn = None

    def _mysql_connect(self):
        logging.info('Connecting to mysql. (retry count=%d)' % (self.mysql_connect.retry.statistics['attempt_number']))
        mysql_engine = create_engine(self.conn_str)
        mysql_conn = mysql_engine.connect()
        logging.info('Connected to mysql successfully with %d attempt(s).' % (self.mysql_connect.retry.statistics['attempt_number']))
        return (mysql_engine, mysql_conn)

    def mysql_connect(self):
        d = retry(
              wait=wait_fixed(self.retry_interval),
              stop=stop_after_attempt(self.retry_count)
            )
        # One of these two should work, depending on how
        # retry is actually defined.
        return d(Connector._mysql_connect)(self)
        # return d(self._mysql_connect)
于 2020-04-02T12:01:53.157 回答
0

除非装饰器（重试）被定义为类的一部分（并且不是静态的），否则您不能引用其中的对象实例。

于 2020-04-02T11:47:45.917 回答
0

如果您不访问@retry装饰器并且阻止对其进行编辑，则可以在类定义之外定义一个变量。或者您可以在设置文件中定义此基本配置，然后将其导入。请看这个：

## etl_connect.py
from sqlalchemy import create_engine
import pymysql
import logging
from tenacity import *


base_config = {
    'retry_count': 3,
    'retry_interval': 30,
    ...
}

class Connector():
    def __init__(self, mode, conn_str):
        self.mode = mode
        self.conn_str = conn_str
        self.engine = None
        self.conn = None

    @retry(wait=wait_fixed(base_config['retry_interval']), stop=stop_after_attempt(base_config['retry_count']))
    def mysql_connect(self):
        logging.info('Connecting to mysql. (retry count=%d)' % (self.mysql_connect.retry.statistics['attempt_number']))
        mysql_engine = create_engine(self.conn_str)
        mysql_conn = mysql_engine.connect()
        logging.info('Connected to mysql successfully with %d attempt(s).' % (self.mysql_connect.retry.statistics['attempt_number']))
        return (mysql_engine, mysql_conn)

或从设置文件导入：

from sqlalchemy import create_engine
import pymysql
import logging
from tenacity import *

from settings import RETRY_COUNT
from settings import RETRY_INTERVAL

我知道这种方式是可靠的，但是这些选项应该在您的设置中定义，并且每次您想要创建类的实例时都不需要传递它们。

于 2020-04-02T12:13:38.003 回答
0

假设您不能轻易地重新定义 tenacityretry装饰器，您可以用您自己的一个引用Connector实例中的值的来包装它。

这就是我的意思：

# Wrapper for tenacity retry decorator
def my_retry(func):
    def wrapped(conn, *args, **kwargs):
        tdecorator = retry(wait=wait_fixed(conn.retry_interval),
                           stop=stop_after_attempt(conn.retry_count))
        decorated = tdecorator(func)
        return decorated(conn, *args, **kwargs)
    return wrapped


class Connector():
    def __init__(self, mode, conn_str, retry_count, retry_interval):
        self.mode = mode
        self.conn_str = conn_str
        self.retry_count = retry_count
        self.retry_interval = retry_interval
        self.engine = None
        self.conn = None

    @my_retry
    def mysql_connect(self):
        logging.info('Connecting to mysql. (retry count=%d)' % (self.mysql_connect.retry.statistics['attempt_number']))
        mysql_engine = create_engine(self.conn_str)
        mysql_conn = mysql_engine.connect()
        logging.info('Connected to mysql successfully with %d attempt(s).' % (self.mysql_connect.retry.statistics['attempt_number']))
        return (mysql_engine, mysql_conn)
于 2020-04-02T13:20:49.217 回答