Decorador de subprocesos [Python]

Question

Estoy tratando de crear un programa simple usando Python socket y biblioteca de subprocesos. Me gustaría automatizar el siguiente procedimiento usando un decorador:

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

el programa está estructurado usando OOP, así que definí una subclase dentro de la principal para contener a todos los decoradores (he leído sobre este método en este artículo: https://medium.com/@vadimpushtaev/decorator-inside-python-class-1e74d23107f6). Por eso tengo una situación 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

Pero cuando trato de correr, me sale el siguiente error: TypeError: function_to_be_threaded() missing one required argument: 'self'. Sospecho que el problema se encuentra en la parte cuando llamo al hilo. Thread (target = function) que de alguna manera no pasa toda la función self.function_to_be_threaded. Por lo tanto, si sabe cómo solucionar este problema, ¿puede decírmelo? Además, ¿puede decirme si hay una manera de implementar un decorador que acepte un argumento que pasará a la clase Thread como args=(arguments_of_the_decorator)?

Muchas gracias por tu tiempo y disculpa mi inglés, sigo practicándolo

Answer 1

Utilice la *argssintaxis para mover los argumentos. En otras palabras, utilice *argspara recopilar todos los argumentos posicionales como una tupla y moverlos 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

ver también ¿Cómo puedo pasar argumentos de una función a otra?