Threading Decorator [Python]

Question

Estou tentando criar um programa simples usando o socket python e a biblioteca de threading. Eu gostaria de automatizar o seguinte procedimento usando um decorador:

t = threading.Thread(target=function, args=(arg1, arg2))
t.start()

o programa é estruturado usando OOP, então eu defini uma subclasse dentro da principal para conter todos os decoradores (eu li sobre este método neste artigo: https://medium.com/@vadimpushtaev/decorator-inside-python-class-1e74d23107f6) Portanto, tenho uma situação como esta:

class Server(object):

    class Decorators(object):

        @classmethod
        def threaded_decorator(cls, function):
            def inner_function():
                function_thread = threading.Thread(target=function)
                function_thread.start()
            return inner_function

    def __init__(self, other_arguments):
        # other code
        pass

    @Decorators.threaded_decorator
    def function_to_be_threaded(self):
        # other code
        pass

Mas quando tento executar, eu recebo o seguinte erro: TypeError: function_to_be_threaded() missing one required argument: 'self'. Suspeito que o problema está localizado na parte quando eu chamo o threading.Thread (target = function) que de alguma forma não passa a função self.function_to_be_threaded inteira. Portanto, se você sabe como consertar isso, por favor, pode me dizer ?. Além disso, você pode me dizer se existe uma maneira de implementar um decorador que aceita argumentos que serão passados ​​para a classe Thread como args=(arguments_of_the_decorator)?

Muito obrigado pelo seu tempo e desculpe meu inglês, ainda estou praticando

Answer 1

Use a *argssintaxe para mover os argumentos. Em outras palavras, use *argspara coletar todos os argumentos posicionais como uma tupla e movê-los threading.Threadcomo args.

import threading
import time
class Server(object):

    class Decorators(object):

        @classmethod
        def threaded_decorator(cls, function):
            def inner_function(*args):
                function_thread = threading.Thread(target=function,args=args)
                function_thread.start()
            return inner_function

    def __init__(self, count,sleep):
        self.count = count
        self.sleep = sleep

    @Decorators.threaded_decorator
    def function_to_be_threaded(self,id):
        for xx in range(self.count):
            time.sleep(self.sleep)
            print("{} ==> {}".format(id,xx))
           

---

>>> Server(6,1).function_to_be_threaded('a')
>>> Server(2,3).function_to_be_threaded('b')

a ==> 0
a ==> 1
a ==> 2
b ==> 0
a ==> 3
a ==> 4
a ==> 5
b ==> 1

veja também Como posso passar argumentos de uma função para outra?