Getting reportOptionalCall error when assigning a variable to a bound method with if statements?

[HunterAP23]

First, here's my code snippet:

def _check_instance(
    var: Any, name: str, types: Union[Type, tuple[Type, ...]], can_be_none=True
) -> None:
    if not isinstance(var, types) and var is None and not can_be_none:
        raise TypeError("'{}' object is not of type {}.".format(name, types))


class amped:
    def __init__(self, log=None):
        ...
        if log is not None and isinstance(log, logging.RootLogger):
            self._log = log
        else:
            self._log = None

   def _log_exception(self, ex):
        if self._log is not None:
            self._log.error(ex)
        raise type(ex)(ex)

    ....

    def create(
            self,
            library: str = "concurrent",
            pool_type: str = "thread",
            group: str = "",
            name: str = "",
            affinity: Union[Sequence[int], Set[int]] = (),
            initializer: Optional[Callable[..., None]] = None,
            initargs: Tuple[Any, ...] = (),
        ):
            """
            Create a pool of threads or processes.
            """
            # Make sure the 'library' argument is a string
            _check_instance(library, "library", str, can_be_none=False)
            # Make sure the 'pool_type' argument is a string
            _check_instance(pool_type, "pool_type", str, can_be_none=False)
    
            # Defines what class to use based on given call to this function
            pool_class = None
    
            # 'multiprocessing' will only support a pool of processes
            # No reason to create a thread pool using the multiprocessing library
            # Incorrect 'pool_type' values will raise an exception
            if library == "multiprocessing":
                if pool_type == "process":
                    pool_class = _amped_process_pool_mp
                else:
                    msg = "'multiprocessing' library only supports 'process' pool_type."
                    self._log_exception(ValueError(msg))
            # 'concurrent' library supports both process and thread pools
            elif library == "concurrent":
                if pool_type == "process":
                    pool_class = _amped_process_pool_cf
                elif pool_type == "thread":
                    pool_class = _amped_thread_pool_cf
                else:
                    msg = "'pool_type' has an incorrect value - it should either be 'process' or 'thread'."
                    self._log_exception(ValueError(msg))
    
            ...
    
            # Create the pool with a given name under the given group with the
            # related data filled in.
            self._pools[group][name] = {}
            self._pools[group][name]["parent"] = psutil.Process().parent()
            self._pools[group][name]["affinity"] = tuple(affinity)
            self._pools[group][name]["initializer"] = initializer
            self._pools[group][name]["initargs"] = tuple(initargs)
    
            # Attempt to create the pool with the given variables
            # Otherwise raise an error
            try:
                self._pools[group][name]["pool"] = pool_class(
                    name=name,
                    affinity=self._pools[group][name]["affinity"],
                    # mp_context=self._mp_ctx,
                    initializer=self._pools[group][name]["initializer"],
                    initargs=self._pools[group][name]["initargs"],
                )
                return self._pools[group][name]["pool"]
            except Exception as e:
                self._log_exception(e)

It might be a lot to read, but the general gist is that this creates a process/thread pool from either multiprocessing.Pool, concurrent.futures.ProcessPoolExecutor, or concurrent.futures.ThreadPoolExecutor

I check that the input arguments for the amped.create method, and then handle the rest accordingly.
Within the amped.create method, I declare pool_class = None as a placeholder, where I later assign it the proper value of a class initializer depending on the given variables. If the user provided incorrect values, then I call log_exception which will raise an exception and the amped.create function will not finish running.
The issue arises later on in the line:

self._pools[group][name]["pool"] = pool_class(...)

where I get the error: Object of type "None" cannot be called (reportOptionalCall).
This error only started appearing when I went from having a single if/else statement to a nested one as shown in the code above.
The old code did not have the library variable and simply looked like this:

if pool_type == "process":
    pool_class = _amped_process_pool_cf
elif pool_type == "thread":
    pool_class = _amped_thread_pool_cf
else:
    msg = "'pool_type' has an incorrect value - it should either be 'process' or 'thread'."
    self._log_exception(ValueError(msg))

So my question is this: is there a different way I should be going about assigning a class initializer to the pool_class with these nested if statements? I wouldn't want to create a class member variable inside the amped class since the amped.create function would have it's own pool_class for each call given what the user input.

[erictraut]

The problem is that you're not handling the case where library contains a value other than "multiprocessing" or "concurrent".

There are a couple of ways you could fix this.

1. Annotate the library parameter with the type Literal["concurrent", "multiprocessing"]. This tells the type checker that it will never contain a value other than these two literal values.
2. Add an else clause that handles the case where library contains an unexpected value and call self._log_exception in that case.