例如,我有一个Django自定义管理命令,它定期(APScheduler + CronTrigger)向Dramatiq发送任务。
为什么下面的代码有不同的函数:
def get_crontab(options):
"""Returns crontab whether from options or settings"""
crontab = options.get("crontab")
if crontab is None:
if not hasattr(settings, "REMOVE_TOO_OLD_CRONTAB"):
raise ImproperlyConfigured("Whether set settings.REMOVE_TOO_OLD_CRONTAB or use --crontab argument")
crontab = settings.REMOVE_TOO_OLD_CRONTAB
return crontab
def add_cron_job(scheduler: BaseScheduler, actor, crontab):
"""Adds cron job which triggers Dramatiq actor"""
module_path = actor.fn.__module__
actor_name = actor.fn.__name__
trigger = CronTrigger.from_crontab(crontab)
job_path = f"{module_path}:{actor_name}.send"
job_name = f"{module_path}.{actor_name}"
scheduler.add_job(job_path, trigger=trigger, name=job_name)
def run_scheduler(scheduler):
"""Runs scheduler in a blocking way"""
def shutdown(signum, frame):
scheduler.shutdown()
signal.signal(signal.SIGINT, shutdown)
signal.signal(signal.SIGTERM, shutdown)
scheduler.start()
class Command(BaseCommand):
help = "Periodically removes too old publications from the RSS feed"
def add_arguments(self, parser: argparse.ArgumentParser):
parser.add_argument("--crontab", type=str)
def handle(self, *args, **options):
scheduler = BlockingScheduler()
add_cron_job(scheduler, tasks.remove_too_old_publications, get_crontab(options))
run_scheduler(scheduler)比有方法的代码更好吗?
class Command(BaseCommand):
help = "Periodically removes too old publications from the RSS feed"
def add_arguments(self, parser: argparse.ArgumentParser):
parser.add_argument("--crontab", type=str)
def get_crontab(self, options):
"""Returns crontab whether from options or settings"""
crontab = options.get("crontab")
if crontab is None:
if not hasattr(settings, "REMOVE_TOO_OLD_CRONTAB"):
raise ImproperlyConfigured(
"Whether set settings.REMOVE_TOO_OLD_CRONTAB or use --crontab argument"
)
crontab = settings.REMOVE_TOO_OLD_CRONTAB
return crontab
def handle(self, *args, **options):
scheduler = BlockingScheduler()
self.add_cron_job(scheduler, tasks.remove_too_old_publications, self.get_crontab(options))
self.run_scheduler(scheduler)
def add_cron_job(self, scheduler: BaseScheduler, actor, crontab):
"""Adds cron job which triggers Dramatiq actor"""
module_path = actor.fn.__module__
actor_name = actor.fn.__name__
trigger = CronTrigger.from_crontab(crontab)
job_path = f"{module_path}:{actor_name}.send"
job_name = f"{module_path}.{actor_name}"
scheduler.add_job(job_path, trigger=trigger, name=job_name)
def run_scheduler(self, scheduler):
"""Runs scheduler in a blocking way"""
def shutdown(signum, frame):
scheduler.shutdown()
signal.signal(signal.SIGINT, shutdown)
signal.signal(signal.SIGTERM, shutdown)
scheduler.start()这段代码只在一个地方使用,不会被重用。
StackOverflow需要更多详细信息,因此:
第二个代码是我最初编写的版本。在那之后,我用Pylint运行了Prospector,除了其他有用的消息外,我还得到了pylint: no-self-use / Method could be a function (col 4)。为了解决这个问题,我重写了代码,如第一个示例所示。但我还是不明白为什么这样更好。
发布于 2021-02-19 22:18:25
至少,在这种情况下,它并不是更好。Pylint会通知您"self“未使用,就像它会通知您变量或导入未使用一样。
修复pylint消息的其他几种选择是在函数中实际使用"self“,或者添加staticmethod (或classmethod)装饰器。两者的示例都在水平线之后。Here are the docs for staticmethod和here's the difference between staticmethod and classmethod。
由于这是一个Django-,而且你可能不会有这个类的多个实例或者继承命令的其他类(比如重载函数),或者可以从类中的函数中受益的东西,所以会选择你认为最易读/最容易改变Django的一个。
为了完整起见,StackExchange Code Review可以进一步了解哪些是最好的,如果有的话。
使用self的示例,主要区别在于调度程序是在__init__中创建的,而不是作为参数传递给使用它的函数:
class Command(BaseCommand):
help = "Periodically removes too old publications from the RSS feed"
def __init__(self):
super().__init__()
self.scheduler = BlockingScheduler()
def add_arguments(self, parser: argparse.ArgumentParser):
parser.add_argument("--crontab", type=str)
def handle(self, *args, **options):
self.add_cron_job(tasks.remove_too_old_publications, self.get_crontab(options))
self.run_scheduler()
# ...
def run_scheduler(self):
"""Runs scheduler in a blocking way"""
def shutdown(signum, frame):
self.scheduler.shutdown()
signal.signal(signal.SIGINT, shutdown)
signal.signal(signal.SIGTERM, shutdown)
self.scheduler.start()使用staticmethod的示例,其中唯一的区别是staticmethod-修饰器和带有装饰器的函数没有自参数:
class Command(BaseCommand):
help = "Periodically removes too old publications from the RSS feed"
def add_arguments(self, parser: argparse.ArgumentParser):
parser.add_argument("--crontab", type=str)
def handle(self, *args, **options):
scheduler = BlockingScheduler()
self.add_cron_job(scheduler, tasks.remove_too_old_publications, self.get_crontab(options))
self.run_scheduler(scheduler)
# ...
@staticmethod
def run_scheduler(scheduler):
"""Runs scheduler in a blocking way"""
def shutdown(signum, frame):
scheduler.shutdown()
signal.signal(signal.SIGINT, shutdown)
signal.signal(signal.SIGTERM, shutdown)
scheduler.start()https://stackoverflow.com/questions/66273419
复制相似问题