Saturday, 1 October 2022

Instantiate threading within a class

I have a class within a class and want to activate threading capabilities in the second class. Essentially, the script below is a reproducible template of my proper project.

When I use @threading I get that showit is not iterable, so the tp.map thinks I do not have a list.

However, when I run:

if __name__ == '__main__':
    tp = ThreadPoolExecutor(5)
    print(tp.map(testit(id_str).test_first, id_int))
    for values in tp.map(testit(id_str).test_first, id_int):
        values

I get no issues, besides that I want the expected output to print out each number in the list. However, I wanted to achieve this within the class.

Something like the following:

from concurrent.futures import ThreadPoolExecutor
from typing import List

id_str = ['1', '2', '3', '4', '5']
id_int = [1, 2, 3, 4, 5]

def threaded(fn, pools=10):
    tp = ThreadPoolExecutor(pools)
    def wrapper(*args):
        return tp.map(fn, *args)  # returns Future object
    return wrapper

class testit:
    def __init__(self, some_list: List[str]) -> None:
        self._ids = some_list
        print(self._ids)

    def test_first(self, some_id: List[int]) -> None:
        print(some_id)

class showit(testit):
    def __init__(self, *args):
        super(showit, self).__init__(*args)
    
    @threaded
    def again(self):
        global id_int
        for values in self.test_first(id_int):
            print(values)

a = showit(id_str)
print(a.again())

Error:

  File "test_1.py", line 32, in <module>
    print(a.again())
  File "test_1.py", line 10, in wrapper
    return tp.map(fn, *args)  # returns Future object
  File "/Users/usr/opt/anaconda3/lib/python3.8/concurrent/futures/_base.py", line 600, in map
    fs = [self.submit(fn, *args) for args in zip(*iterables)]
TypeError: 'showit' object is not iterable

Expected output:

1
2
3
4
5



from Instantiate threading within a class

No comments:

Post a Comment