传递自身 class 参数 python
Pass self class parameter python
这是我的 class 充满静态函数的定义。我想在 "sendLog" 函数中使用所有这些函数,该函数以时间间隔(此处为 10 秒)调用自己。当我 运行 这个解释器告诉我 "TypeError: sendLog() takes at least 5 arguments (0 given)"
但是,如果我输入相同的参数,我将需要一次又一次地定义 sendLog,因为它会重复调用自己。我知道这不是办法,但无法弄清楚。
class AccessLog:
@staticmethod
def backupAccessLog(target, source):
newfile = os.path.splitext(source)[0] + "_" + time.strftime("%Y%m%d-%H%M%S") + os.path.splitext(source)[1]
copyfile(source,newfile)
shutil.move(newfile,target)
@staticmethod
def emptyAccessLog(filename):
open(filename, 'w').close()
@staticmethod
def postLogstoElastic():
fileLogs = open("example.log", "rw+")
fileBackups = open("logs_of_accesslog.log","rw+")
lines = fileLogs.read().splitlines()
logging.basicConfig(format='%(asctime)s>>>%(message)s',filename='logs_exceptions.log',level=logging.DEBUG)
es = Elasticsearch(['http://localhost:9200/'], verify_certs=True)
#es.create(index="index_log23June", doc_type="type_log23June")
es.indices.create(index='index_log23June', ignore=400)
i=0
for item in lines:
try:
i+=1
if bool(item):
es.index(index="index_log23June",doc_type="type_log23June", body={"Log":item})
else:
print "a speace line ignored. at line number:", i
raise ValueError('Error occurred on this line: ', i)
print "lines[",i,"]:",item,"\n"
except ValueError as err:
logging.error(err.args)
@staticmethod
def sendLog(interval, worker_functions, iterations=1):
def call_worker_functions():
for f in worker_functions:
f() #ERROR: Msg: 'NoneType' object is not callable
for i in range(iterations):
threading.Timer(interval * i, call_worker_functions).start()
我想用这一行调用这个方法:
try:
AccessLog.AccessLog.sendLog(
interval=10,
worker_functions=(
AccessLog.AccessLog.backupAccessLog("logbackups","example.log"),
AccessLog.AccessLog.emptyAccessLog("example.log"),
AccessLog.AccessLog.postLogstoElastic()
),
iterations=999
)
except ValueError as err:
logging.error(err.args)
"TypeError: sendLog() takes at least 5 arguments (0 given)" 看起来很正常,但我该如何处理?
您是否尝试过将@staticmethod 设置为与该函数处于同一级别?
显然您希望 sendLog()
每 10 秒左右调用一次辅助函数。
这是一个简单的方法:
class AccessLog:
@staticmethod
def sendLog(interval, worker_functions, iterations=1):
def call_worker_functions():
for f in worker_functions:
f(*worker_functions[f])
for i in range(iterations):
threading.Timer(interval * i, call_worker_functions).start()
现在像这样使用它:
AccessLog.AccessLog.sendLog(
interval=10,
worker_functions={
AccessLog.AccessLog.backupAccessLog: ("logbackups", "example.log"),
AccessLog.AccessLog.emptyAccessLog: ("example.log",),
AccessLog.AccessLog.postLogstoElastic: ()
),
iterations=999
)
这只是众多方法中的一种,但不需要像您那样将函数作为其自身的参数传递。
这是我的 class 充满静态函数的定义。我想在 "sendLog" 函数中使用所有这些函数,该函数以时间间隔(此处为 10 秒)调用自己。当我 运行 这个解释器告诉我 "TypeError: sendLog() takes at least 5 arguments (0 given)" 但是,如果我输入相同的参数,我将需要一次又一次地定义 sendLog,因为它会重复调用自己。我知道这不是办法,但无法弄清楚。
class AccessLog:
@staticmethod
def backupAccessLog(target, source):
newfile = os.path.splitext(source)[0] + "_" + time.strftime("%Y%m%d-%H%M%S") + os.path.splitext(source)[1]
copyfile(source,newfile)
shutil.move(newfile,target)
@staticmethod
def emptyAccessLog(filename):
open(filename, 'w').close()
@staticmethod
def postLogstoElastic():
fileLogs = open("example.log", "rw+")
fileBackups = open("logs_of_accesslog.log","rw+")
lines = fileLogs.read().splitlines()
logging.basicConfig(format='%(asctime)s>>>%(message)s',filename='logs_exceptions.log',level=logging.DEBUG)
es = Elasticsearch(['http://localhost:9200/'], verify_certs=True)
#es.create(index="index_log23June", doc_type="type_log23June")
es.indices.create(index='index_log23June', ignore=400)
i=0
for item in lines:
try:
i+=1
if bool(item):
es.index(index="index_log23June",doc_type="type_log23June", body={"Log":item})
else:
print "a speace line ignored. at line number:", i
raise ValueError('Error occurred on this line: ', i)
print "lines[",i,"]:",item,"\n"
except ValueError as err:
logging.error(err.args)
@staticmethod
def sendLog(interval, worker_functions, iterations=1):
def call_worker_functions():
for f in worker_functions:
f() #ERROR: Msg: 'NoneType' object is not callable
for i in range(iterations):
threading.Timer(interval * i, call_worker_functions).start()
我想用这一行调用这个方法:
try:
AccessLog.AccessLog.sendLog(
interval=10,
worker_functions=(
AccessLog.AccessLog.backupAccessLog("logbackups","example.log"),
AccessLog.AccessLog.emptyAccessLog("example.log"),
AccessLog.AccessLog.postLogstoElastic()
),
iterations=999
)
except ValueError as err:
logging.error(err.args)
"TypeError: sendLog() takes at least 5 arguments (0 given)" 看起来很正常,但我该如何处理?
您是否尝试过将@staticmethod 设置为与该函数处于同一级别?
显然您希望 sendLog()
每 10 秒左右调用一次辅助函数。
这是一个简单的方法:
class AccessLog:
@staticmethod
def sendLog(interval, worker_functions, iterations=1):
def call_worker_functions():
for f in worker_functions:
f(*worker_functions[f])
for i in range(iterations):
threading.Timer(interval * i, call_worker_functions).start()
现在像这样使用它:
AccessLog.AccessLog.sendLog(
interval=10,
worker_functions={
AccessLog.AccessLog.backupAccessLog: ("logbackups", "example.log"),
AccessLog.AccessLog.emptyAccessLog: ("example.log",),
AccessLog.AccessLog.postLogstoElastic: ()
),
iterations=999
)
这只是众多方法中的一种,但不需要像您那样将函数作为其自身的参数传递。