diff --git a/scrapy/commands/parse.py b/scrapy/commands/parse.py
index 2453c0d39..f916a3e75 100644
--- a/scrapy/commands/parse.py
+++ b/scrapy/commands/parse.py
@@ -140,13 +140,13 @@ class Command(BaseRunSpiderCommand):
 
     @overload
     def iterate_spider_output(
-        self, result: Union[AsyncGenerator, CoroutineType]
-    ) -> Deferred: ...
+        self, result: Union[AsyncGenerator[_T, None], CoroutineType[Any, Any, _T]]
+    ) -> Deferred[_T]: ...
 
     @overload
-    def iterate_spider_output(self, result: _T) -> Iterable: ...
+    def iterate_spider_output(self, result: _T) -> Iterable[Any]: ...
 
-    def iterate_spider_output(self, result: Any) -> Union[Iterable, Deferred]:
+    def iterate_spider_output(self, result: Any) -> Union[Iterable[Any], Deferred]:
         if inspect.isasyncgen(result):
             d = deferred_from_coro(
                 collect_asyncgen(aiter_errback(result, self.handle_exception))
diff --git a/scrapy/core/engine.py b/scrapy/core/engine.py
index b342ad7a3..dededf99d 100644
--- a/scrapy/core/engine.py
+++ b/scrapy/core/engine.py
@@ -372,7 +372,10 @@ class ExecutionEngine:
 
     @inlineCallbacks
     def open_spider(
-        self, spider: Spider, start_requests: Iterable = (), close_if_idle: bool = True
+        self,
+        spider: Spider,
+        start_requests: Iterable[Request] = (),
+        close_if_idle: bool = True,
     ) -> Generator[Deferred, Any, None]:
         if self.slot is not None:
             raise RuntimeError(f"No free spider slot when opening {spider.name!r}")
diff --git a/scrapy/core/scraper.py b/scrapy/core/scraper.py
index 566e6628b..3b7492838 100644
--- a/scrapy/core/scraper.py
+++ b/scrapy/core/scraper.py
@@ -16,6 +16,7 @@ from typing import (
     Set,
     Tuple,
     Type,
+    TypeVar,
     Union,
     cast,
 )
@@ -47,6 +48,7 @@ if TYPE_CHECKING:
     from scrapy.crawler import Crawler
 
 
+_T = TypeVar("_T")
 QueueTuple = Tuple[Union[Response, Failure], Request, Deferred]
 
 
@@ -256,14 +258,14 @@ class Scraper:
 
     def handle_spider_output(
         self,
-        result: Union[Iterable, AsyncIterable],
+        result: Union[Iterable[_T], AsyncIterable[_T]],
         request: Request,
         response: Response,
         spider: Spider,
     ) -> Deferred:
         if not result:
             return defer_succeed(None)
-        it: Union[Iterable, AsyncIterable]
+        it: Union[Iterable[_T], AsyncIterable[_T]]
         if isinstance(result, AsyncIterable):
             it = aiter_errback(
                 result, self.handle_spider_error, request, response, spider
diff --git a/scrapy/core/spidermw.py b/scrapy/core/spidermw.py
index 2cef2e1dd..cb1a93a68 100644
--- a/scrapy/core/spidermw.py
+++ b/scrapy/core/spidermw.py
@@ -9,7 +9,6 @@ from inspect import isasyncgenfunction, iscoroutine
 from itertools import islice
 from typing import (
     Any,
-    AsyncGenerator,
     AsyncIterable,
     Callable,
     Generator,
@@ -17,6 +16,7 @@ from typing import (
     List,
     Optional,
     Tuple,
+    TypeVar,
     Union,
     cast,
 )
@@ -42,6 +42,7 @@ from scrapy.utils.python import MutableAsyncChain, MutableChain
 logger = logging.getLogger(__name__)
 
 
+_T = TypeVar("_T")
 ScrapeFunc = Callable[[Union[Response, Failure], Request, Spider], Any]
 
 
@@ -98,31 +99,39 @@ class SpiderMiddlewareManager(MiddlewareManager):
         self,
         response: Response,
         spider: Spider,
-        iterable: Union[Iterable, AsyncIterable],
+        iterable: Union[Iterable[_T], AsyncIterable[_T]],
         exception_processor_index: int,
-        recover_to: Union[MutableChain, MutableAsyncChain],
-    ) -> Union[Generator, AsyncGenerator]:
-        def process_sync(iterable: Iterable) -> Generator:
+        recover_to: Union[MutableChain[_T], MutableAsyncChain[_T]],
+    ) -> Union[Iterable[_T], AsyncIterable[_T]]:
+        def process_sync(iterable: Iterable[_T]) -> Iterable[_T]:
             try:
                 yield from iterable
             except Exception as ex:
-                exception_result = self._process_spider_exception(
-                    response, spider, Failure(ex), exception_processor_index
+                exception_result = cast(
+                    Union[Failure, MutableChain[_T]],
+                    self._process_spider_exception(
+                        response, spider, Failure(ex), exception_processor_index
+                    ),
                 )
                 if isinstance(exception_result, Failure):
                     raise
+                assert isinstance(recover_to, MutableChain)
                 recover_to.extend(exception_result)
 
-        async def process_async(iterable: AsyncIterable) -> AsyncGenerator:
+        async def process_async(iterable: AsyncIterable[_T]) -> AsyncIterable[_T]:
             try:
                 async for r in iterable:
                     yield r
             except Exception as ex:
-                exception_result = self._process_spider_exception(
-                    response, spider, Failure(ex), exception_processor_index
+                exception_result = cast(
+                    Union[Failure, MutableAsyncChain[_T]],
+                    self._process_spider_exception(
+                        response, spider, Failure(ex), exception_processor_index
+                    ),
                 )
                 if isinstance(exception_result, Failure):
                     raise
+                assert isinstance(recover_to, MutableAsyncChain)
                 recover_to.extend(exception_result)
 
         if isinstance(iterable, AsyncIterable):
@@ -135,7 +144,7 @@ class SpiderMiddlewareManager(MiddlewareManager):
         spider: Spider,
         _failure: Failure,
         start_index: int = 0,
-    ) -> Union[Failure, MutableChain]:
+    ) -> Union[Failure, MutableChain[_T], MutableAsyncChain[_T]]:
         exception = _failure.value
         # don't handle _InvalidOutput exception
         if isinstance(exception, _InvalidOutput):
@@ -151,14 +160,18 @@ class SpiderMiddlewareManager(MiddlewareManager):
             if _isiterable(result):
                 # stop exception handling by handing control over to the
                 # process_spider_output chain if an iterable has been returned
-                dfd: Deferred = self._process_spider_output(
-                    response, spider, result, method_index + 1
+                dfd: Deferred[Union[MutableChain[_T], MutableAsyncChain[_T]]] = (
+                    self._process_spider_output(
+                        response, spider, result, method_index + 1
+                    )
                 )
                 # _process_spider_output() returns a Deferred only because of downgrading so this can be
                 # simplified when downgrading is removed.
                 if dfd.called:
                     # the result is available immediately if _process_spider_output didn't do downgrading
-                    return cast(MutableChain, dfd.result)
+                    return cast(
+                        Union[MutableChain[_T], MutableAsyncChain[_T]], dfd.result
+                    )
                 # we forbid waiting here because otherwise we would need to return a deferred from
                 # _process_spider_exception too, which complicates the architecture
                 msg = f"Async iterable returned from {method.__qualname__} cannot be downgraded"
@@ -181,12 +194,12 @@ class SpiderMiddlewareManager(MiddlewareManager):
         self,
         response: Response,
         spider: Spider,
-        result: Union[Iterable, AsyncIterable],
+        result: Union[Iterable[_T], AsyncIterable[_T]],
         start_index: int = 0,
-    ) -> Generator[Deferred, Any, Union[MutableChain, MutableAsyncChain]]:
+    ) -> Generator[Deferred[Any], Any, Union[MutableChain[_T], MutableAsyncChain[_T]]]:
         # items in this iterable do not need to go through the process_spider_output
         # chain, they went through it already from the process_spider_exception method
-        recovered: Union[MutableChain, MutableAsyncChain]
+        recovered: Union[MutableChain[_T], MutableAsyncChain[_T]]
         last_result_is_async = isinstance(result, AsyncIterable)
         if last_result_is_async:
             recovered = MutableAsyncChain()
@@ -237,7 +250,9 @@ class SpiderMiddlewareManager(MiddlewareManager):
                 # might fail directly if the output value is not a generator
                 result = method(response=response, result=result, spider=spider)
             except Exception as ex:
-                exception_result = self._process_spider_exception(
+                exception_result: Union[
+                    Failure, MutableChain[_T], MutableAsyncChain[_T]
+                ] = self._process_spider_exception(
                     response, spider, Failure(ex), method_index + 1
                 )
                 if isinstance(exception_result, Failure):
@@ -267,9 +282,12 @@ class SpiderMiddlewareManager(MiddlewareManager):
         return MutableChain(result, recovered)  # type: ignore[arg-type]
 
     async def _process_callback_output(
-        self, response: Response, spider: Spider, result: Union[Iterable, AsyncIterable]
-    ) -> Union[MutableChain, MutableAsyncChain]:
-        recovered: Union[MutableChain, MutableAsyncChain]
+        self,
+        response: Response,
+        spider: Spider,
+        result: Union[Iterable[_T], AsyncIterable[_T]],
+    ) -> Union[MutableChain[_T], MutableAsyncChain[_T]]:
+        recovered: Union[MutableChain[_T], MutableAsyncChain[_T]]
         if isinstance(result, AsyncIterable):
             recovered = MutableAsyncChain()
         else:
@@ -293,14 +311,16 @@ class SpiderMiddlewareManager(MiddlewareManager):
         spider: Spider,
     ) -> Deferred:
         async def process_callback_output(
-            result: Union[Iterable, AsyncIterable]
-        ) -> Union[MutableChain, MutableAsyncChain]:
+            result: Union[Iterable[_T], AsyncIterable[_T]]
+        ) -> Union[MutableChain[_T], MutableAsyncChain[_T]]:
             return await self._process_callback_output(response, spider, result)
 
-        def process_spider_exception(_failure: Failure) -> Union[Failure, MutableChain]:
+        def process_spider_exception(
+            _failure: Failure,
+        ) -> Union[Failure, MutableChain[_T], MutableAsyncChain[_T]]:
             return self._process_spider_exception(response, spider, _failure)
 
-        dfd = mustbe_deferred(
+        dfd: Deferred = mustbe_deferred(
             self._process_spider_input, scrape_func, response, request, spider
         )
         dfd.addCallback(deferred_f_from_coro_f(process_callback_output))
diff --git a/scrapy/http/response/__init__.py b/scrapy/http/response/__init__.py
index 166c4de97..daf193f59 100644
--- a/scrapy/http/response/__init__.py
+++ b/scrapy/http/response/__init__.py
@@ -14,7 +14,6 @@ from typing import (
     AnyStr,
     Callable,
     Dict,
-    Generator,
     Iterable,
     List,
     Mapping,
@@ -242,7 +241,7 @@ class Response(object_ref):
         errback: Optional[Callable] = None,
         cb_kwargs: Optional[Dict[str, Any]] = None,
         flags: Optional[List[str]] = None,
-    ) -> Generator[Request, None, None]:
+    ) -> Iterable[Request]:
         """
         .. versionadded:: 2.0
 
diff --git a/scrapy/http/response/text.py b/scrapy/http/response/text.py
index 44c36b682..df4d90829 100644
--- a/scrapy/http/response/text.py
+++ b/scrapy/http/response/text.py
@@ -15,7 +15,6 @@ from typing import (
     AnyStr,
     Callable,
     Dict,
-    Generator,
     Iterable,
     List,
     Mapping,
@@ -246,7 +245,7 @@ class TextResponse(Response):
         flags: Optional[List[str]] = None,
         css: Optional[str] = None,
         xpath: Optional[str] = None,
-    ) -> Generator[Request, None, None]:
+    ) -> Iterable[Request]:
         """
         A generator that produces :class:`~.Request` instances to follow all
         links in ``urls``. It accepts the same arguments as the :class:`~.Request`'s
diff --git a/scrapy/utils/iterators.py b/scrapy/utils/iterators.py
index cd6e9d04e..41a842386 100644
--- a/scrapy/utils/iterators.py
+++ b/scrapy/utils/iterators.py
@@ -6,8 +6,7 @@ from typing import (
     Any,
     Callable,
     Dict,
-    Generator,
-    Iterable,
+    Iterator,
     List,
     Literal,
     Optional,
@@ -22,14 +21,12 @@ from lxml import etree  # nosec
 from scrapy.exceptions import ScrapyDeprecationWarning
 from scrapy.http import Response, TextResponse
 from scrapy.selector import Selector
-from scrapy.utils.python import re_rsearch, to_unicode
+from scrapy.utils.python import re_rsearch
 
 logger = logging.getLogger(__name__)
 
 
-def xmliter(
-    obj: Union[Response, str, bytes], nodename: str
-) -> Generator[Selector, Any, None]:
+def xmliter(obj: Union[Response, str, bytes], nodename: str) -> Iterator[Selector]:
     """Return a iterator of Selector's over all nodes of a XML document,
        given the name of the node to iterate. Useful for parsing XML feeds.
 
@@ -90,7 +87,7 @@ def xmliter_lxml(
     nodename: str,
     namespace: Optional[str] = None,
     prefix: str = "x",
-) -> Generator[Selector, Any, None]:
+) -> Iterator[Selector]:
     reader = _StreamReader(obj)
     tag = f"{{{namespace}}}{nodename}" if namespace else nodename
     iterable = etree.iterparse(
@@ -168,7 +165,7 @@ def csviter(
     headers: Optional[List[str]] = None,
     encoding: Optional[str] = None,
     quotechar: Optional[str] = None,
-) -> Generator[Dict[str, str], Any, None]:
+) -> Iterator[Dict[str, str]]:
     """Returns an iterator of dictionaries from the given csv object
 
     obj can be:
@@ -184,10 +181,13 @@ def csviter(
     quotechar is the character used to enclosure fields on the given obj.
     """
 
-    encoding = obj.encoding if isinstance(obj, TextResponse) else encoding or "utf-8"
-
-    def row_to_unicode(row_: Iterable) -> List[str]:
-        return [to_unicode(field, encoding) for field in row_]
+    if encoding is not None:
+        warn(
+            "The encoding argument of csviter() is ignored and will be removed"
+            " in a future Scrapy version.",
+            category=ScrapyDeprecationWarning,
+            stacklevel=2,
+        )
 
     lines = StringIO(_body_or_str(obj, unicode=True))
 
@@ -200,13 +200,11 @@ def csviter(
 
     if not headers:
         try:
-            row = next(csv_r)
+            headers = next(csv_r)
         except StopIteration:
             return
-        headers = row_to_unicode(row)
 
     for row in csv_r:
-        row = row_to_unicode(row)
         if len(row) != len(headers):
             logger.warning(
                 "ignoring row %(csvlnum)d (length: %(csvrow)d, "
diff --git a/scrapy/utils/misc.py b/scrapy/utils/misc.py
index 49f36de2d..3d11c1035 100644
--- a/scrapy/utils/misc.py
+++ b/scrapy/utils/misc.py
@@ -20,8 +20,8 @@ from typing import (
     Any,
     Callable,
     Deque,
-    Generator,
     Iterable,
+    Iterator,
     List,
     Optional,
     Type,
@@ -227,7 +227,7 @@ def build_from_settings(
 
 
 @contextmanager
-def set_environ(**kwargs: str) -> Generator[None, Any, None]:
+def set_environ(**kwargs: str) -> Iterator[None]:
     """Temporarily set environment variables inside the context manager and
     fully restore previous environment afterwards
     """
@@ -244,7 +244,7 @@ def set_environ(**kwargs: str) -> Generator[None, Any, None]:
                 os.environ[k] = v
 
 
-def walk_callable(node: ast.AST) -> Generator[ast.AST, Any, None]:
+def walk_callable(node: ast.AST) -> Iterable[ast.AST]:
     """Similar to ``ast.walk``, but walks only function body and skips nested
     functions defined within the node.
     """
diff --git a/scrapy/utils/python.py b/scrapy/utils/python.py
index 37a84a350..059d8e04d 100644
--- a/scrapy/utils/python.py
+++ b/scrapy/utils/python.py
@@ -15,12 +15,10 @@ from itertools import chain
 from typing import (
     TYPE_CHECKING,
     Any,
-    AsyncGenerator,
     AsyncIterable,
     AsyncIterator,
     Callable,
     Dict,
-    Generator,
     Iterable,
     Iterator,
     List,
@@ -163,7 +161,7 @@ def re_rsearch(
     the start position of the match, and the ending (regarding the entire text).
     """
 
-    def _chunk_iter() -> Generator[Tuple[str, int], Any, None]:
+    def _chunk_iter() -> Iterable[Tuple[str, int]]:
         offset = len(text)
         while True:
             offset -= chunk_size * 1024
@@ -351,43 +349,45 @@ else:
         gc.collect()
 
 
-class MutableChain(Iterable):
+class MutableChain(Iterable[_T]):
     """
     Thin wrapper around itertools.chain, allowing to add iterables "in-place"
     """
 
-    def __init__(self, *args: Iterable):
-        self.data = chain.from_iterable(args)
+    def __init__(self, *args: Iterable[_T]):
+        self.data: Iterator[_T] = chain.from_iterable(args)
 
-    def extend(self, *iterables: Iterable) -> None:
+    def extend(self, *iterables: Iterable[_T]) -> None:
         self.data = chain(self.data, chain.from_iterable(iterables))
 
-    def __iter__(self) -> Iterator:
+    def __iter__(self) -> Iterator[_T]:
         return self
 
-    def __next__(self) -> Any:
+    def __next__(self) -> _T:
         return next(self.data)
 
 
-async def _async_chain(*iterables: Union[Iterable, AsyncIterable]) -> AsyncGenerator:
+async def _async_chain(
+    *iterables: Union[Iterable[_T], AsyncIterable[_T]]
+) -> AsyncIterator[_T]:
     for it in iterables:
         async for o in as_async_generator(it):
             yield o
 
 
-class MutableAsyncChain(AsyncIterable):
+class MutableAsyncChain(AsyncIterable[_T]):
     """
     Similar to MutableChain but for async iterables
     """
 
-    def __init__(self, *args: Union[Iterable, AsyncIterable]):
-        self.data = _async_chain(*args)
+    def __init__(self, *args: Union[Iterable[_T], AsyncIterable[_T]]):
+        self.data: AsyncIterator[_T] = _async_chain(*args)
 
-    def extend(self, *iterables: Union[Iterable, AsyncIterable]) -> None:
+    def extend(self, *iterables: Union[Iterable[_T], AsyncIterable[_T]]) -> None:
         self.data = _async_chain(self.data, _async_chain(*iterables))
 
-    def __aiter__(self) -> AsyncIterator:
+    def __aiter__(self) -> AsyncIterator[_T]:
         return self
 
-    async def __anext__(self) -> Any:
+    async def __anext__(self) -> _T:
         return await self.data.__anext__()
diff --git a/scrapy/utils/request.py b/scrapy/utils/request.py
index 42a6537a8..45b8008f4 100644
--- a/scrapy/utils/request.py
+++ b/scrapy/utils/request.py
@@ -12,7 +12,6 @@ from typing import (
     TYPE_CHECKING,
     Any,
     Dict,
-    Generator,
     Iterable,
     List,
     Optional,
@@ -40,9 +39,7 @@ if TYPE_CHECKING:
     from scrapy.crawler import Crawler
 
 
-def _serialize_headers(
-    headers: Iterable[bytes], request: Request
-) -> Generator[bytes, Any, None]:
+def _serialize_headers(headers: Iterable[bytes], request: Request) -> Iterable[bytes]:
     for header in headers:
         if header in request.headers:
             yield header
diff --git a/scrapy/utils/sitemap.py b/scrapy/utils/sitemap.py
index cf429043d..7a91afe59 100644
--- a/scrapy/utils/sitemap.py
+++ b/scrapy/utils/sitemap.py
@@ -5,7 +5,7 @@ Note: The main purpose of this module is to provide support for the
 SitemapSpider, its API is subject to change without notice.
 """
 
-from typing import Any, Dict, Generator, Iterator, Optional, Union
+from typing import Any, Dict, Iterable, Iterator, Optional, Union
 from urllib.parse import urljoin
 
 import lxml.etree  # nosec
@@ -42,7 +42,7 @@ class Sitemap:
 
 def sitemap_urls_from_robots(
     robots_text: str, base_url: Optional[str] = None
-) -> Generator[str, Any, None]:
+) -> Iterable[str]:
     """Return an iterator over all sitemap urls contained in the given
     robots.txt file
     """
diff --git a/scrapy/utils/spider.py b/scrapy/utils/spider.py
index cbbb01d85..b05135c04 100644
--- a/scrapy/utils/spider.py
+++ b/scrapy/utils/spider.py
@@ -7,7 +7,6 @@ from typing import (
     TYPE_CHECKING,
     Any,
     AsyncGenerator,
-    Generator,
     Iterable,
     Literal,
     Optional,
@@ -34,18 +33,20 @@ _T = TypeVar("_T")
 
 # https://stackoverflow.com/questions/60222982
 @overload
-def iterate_spider_output(result: AsyncGenerator) -> AsyncGenerator: ...  # type: ignore[overload-overlap]
+def iterate_spider_output(result: AsyncGenerator[_T, None]) -> AsyncGenerator[_T, None]: ...  # type: ignore[overload-overlap]
 
 
 @overload
-def iterate_spider_output(result: CoroutineType) -> Deferred: ...
+def iterate_spider_output(result: CoroutineType[Any, Any, _T]) -> Deferred[_T]: ...
 
 
 @overload
-def iterate_spider_output(result: _T) -> Iterable: ...
+def iterate_spider_output(result: _T) -> Iterable[Any]: ...
 
 
-def iterate_spider_output(result: Any) -> Union[Iterable, AsyncGenerator, Deferred]:
+def iterate_spider_output(
+    result: Any,
+) -> Union[Iterable[Any], AsyncGenerator[_T, None], Deferred[_T]]:
     if inspect.isasyncgen(result):
         return result
     if inspect.iscoroutine(result):
@@ -55,7 +56,7 @@ def iterate_spider_output(result: Any) -> Union[Iterable, AsyncGenerator, Deferr
     return arg_to_iter(deferred_from_coro(result))
 
 
-def iter_spider_classes(module: ModuleType) -> Generator[Type[Spider], Any, None]:
+def iter_spider_classes(module: ModuleType) -> Iterable[Type[Spider]]:
     """Return an iterator over all spider classes defined in the given module
     that can be instantiated (i.e. which have name)
     """
diff --git a/tests/test_commands.py b/tests/test_commands.py
index b9d468c66..857a56b73 100644
--- a/tests/test_commands.py
+++ b/tests/test_commands.py
@@ -13,7 +13,7 @@ from shutil import copytree, rmtree
 from stat import S_IWRITE as ANYONE_WRITE_PERMISSION
 from tempfile import TemporaryFile, mkdtemp
 from threading import Timer
-from typing import Dict, Generator, Optional, Union
+from typing import Dict, Iterator, Optional, Union
 from unittest import skipIf
 
 from pytest import mark
@@ -674,7 +674,7 @@ class BadSpider(scrapy.Spider):
         """
 
     @contextmanager
-    def _create_file(self, content, name=None) -> Generator[str, None, None]:
+    def _create_file(self, content, name=None) -> Iterator[str]:
         tmpdir = Path(self.mktemp())
         tmpdir.mkdir()
         if name: