Repository: aio-libs/aiojobs

Issue:
Why it print exception stack twice?
Code example
```
import asyncio

import aiojobs


async def f():
    try:
        await asyncio.sleep(2)
    except asyncio.CancelledError:
        print('canceled')
    else:
        print('not canceled')


async def err():
    raise Exception


async def main():
    scheduler = await aiojobs.create_scheduler(close_timeout=None)

    await scheduler.spawn(f())
    await scheduler.spawn(err())

    try:
        await scheduler
    except:
        print('exception occurred, cancel')
        await scheduler.close()
    else:
        print('all good')


if __name__ == '__main__':
    asyncio.run(main())
```
Result
```
exception occurred, cancel
Job processing failed
job: <Job closed coro=<<coroutine object err at 0x000001E5768B1D40>>>
Traceback (most recent call last):
  File "C:\Users\Asus\.virtualenvs\test-2uqMFfYk\lib\site-packages\aiojobs\_job.py", line 89, in _close
    await self._task
  File "C:/Users/Asus/Documents/Projects/python/test/sub.py", line 16, in err
    raise Exception
Exception
Job processing failed
job: <Job closed coro=<<coroutine object err at 0x000001E5768B1D40>>>
Traceback (most recent call last):
  File "C:\Users\Asus\.virtualenvs\test-2uqMFfYk\lib\site-packages\aiojobs\_job.py", line 89, in _close
    await self._task
  File "C:/Users/Asus/Documents/Projects/python/test/sub.py", line 16, in err
    raise Exception
Exception
cancelled
```
Why it print stack trace twice?

Discussion:

Patch:
diff --git a/README.rst b/README.rst
index 0344d8d..7b49629 100644
--- a/README.rst
+++ b/README.rst
@@ -39,7 +39,7 @@ Usage example
        await asyncio.sleep(timeout)
 
    async def main():
-       scheduler = await aiojobs.create_scheduler()
+       scheduler = aiojobs.Scheduler()
        for i in range(100):
            # spawn jobs
            await scheduler.spawn(coro(i/10))
diff --git a/aiojobs/__init__.py b/aiojobs/__init__.py
index 9fd89eb..1c3bd82 100644
--- a/aiojobs/__init__.py
+++ b/aiojobs/__init__.py
@@ -4,8 +4,10 @@ The library gives controlled way for scheduling background tasks for
 asyncio applications.
 
 """
+import warnings
 from typing import Optional
 
+from ._job import Job
 from ._scheduler import ExceptionHandler, Scheduler
 
 __version__ = "1.0.0"
@@ -18,6 +20,8 @@ async def create_scheduler(
     pending_limit: int = 10000,
     exception_handler: Optional[ExceptionHandler] = None
 ) -> Scheduler:
+    warnings.warn("Scheduler can now be instantiated directly.", DeprecationWarning)
+
     if exception_handler is not None and not callable(exception_handler):
         raise TypeError(
             "A callable object or None is expected, "
@@ -31,4 +35,4 @@ async def create_scheduler(
     )
 
 
-__all__ = ("Scheduler", "create_scheduler")
+__all__ = ("Job", "Scheduler", "create_scheduler")
diff --git a/aiojobs/_job.py b/aiojobs/_job.py
index f705360..6aded80 100644
--- a/aiojobs/_job.py
+++ b/aiojobs/_job.py
@@ -112,10 +112,9 @@ class Job(Generic[_T]):
             # there's no timeout. self._scheduler will now be None though.
             assert scheduler is not None
             scheduler.call_exception_handler(context)
-        except Exception as exc:
+        except Exception:
             if self._explicit:
                 raise
-            self._report_exception(exc)
 
     def _start(self) -> None:
         assert self._task is None
diff --git a/aiojobs/_scheduler.py b/aiojobs/_scheduler.py
index 2df4ff2..331f60c 100644
--- a/aiojobs/_scheduler.py
+++ b/aiojobs/_scheduler.py
@@ -21,11 +21,17 @@ class Scheduler(Collection[Job[object]]):
     def __init__(
         self,
         *,
-        close_timeout: Optional[float],
-        limit: Optional[int],
-        pending_limit: int,
-        exception_handler: Optional[ExceptionHandler],
+        close_timeout: Optional[float] = 0.1,
+        limit: Optional[int] = 100,
+        pending_limit: int = 10000,
+        exception_handler: Optional[ExceptionHandler] = None,
     ):
+        if exception_handler is not None and not callable(exception_handler):
+            raise TypeError(
+                "A callable object or None is expected, "
+                "got {!r}".format(exception_handler)
+            )
+
         self._jobs: Set[Job[object]] = set()
         self._close_timeout = close_timeout
         self._limit = limit
diff --git a/aiojobs/aiohttp.py b/aiojobs/aiohttp.py
index 0b4cdfa..813cecc 100644
--- a/aiojobs/aiohttp.py
+++ b/aiojobs/aiohttp.py
@@ -12,7 +12,6 @@ from typing import (
 
 from aiohttp import web
 
-from . import create_scheduler
 from ._job import Job
 from ._scheduler import Scheduler
 
@@ -63,7 +62,7 @@ def atomic(
 
 def setup(app: web.Application, **kwargs: Any) -> None:
     async def cleanup_context(app: web.Application) -> AsyncIterator[None]:
-        app["AIOJOBS_SCHEDULER"] = scheduler = await create_scheduler(**kwargs)
+        app["AIOJOBS_SCHEDULER"] = scheduler = Scheduler(**kwargs)
         yield
         await scheduler.close()
 
diff --git a/docs/api.rst b/docs/api.rst
index f4bfc5c..c2018c8 100644
--- a/docs/api.rst
+++ b/docs/api.rst
@@ -8,20 +8,30 @@ API
 .. currentmodule:: aiojobs
 
 
-Instantiation
--------------
+Scheduler
+---------
+
+.. class:: Scheduler(*, close_timeout=0.1, limit=100, \
+                     pending_limit=10000, exception_handler=None)
+
+   A container for managed jobs.
+
+   Jobs are created by :meth:`spawn()`.
 
-.. cofunction:: create_scheduler(*, close_timeout=0.1, limit=100, \
-                                 pending_limit=10000, \
-                                 exception_handler=None)
+   :meth:`close` should be used for finishing all scheduled jobs.
 
-   Create a new :class:`Scheduler`.
+   The class implements :class:`collections.abc.Collection` contract,
+   jobs could be iterated etc.: ``len(scheduler)``, ``for job in
+   scheduler``, ``job in scheduler`` operations are supported.
+
+   Class must be instantiated within a running event loop (e.g. in an
+   ``async`` function).
 
    * *close_timeout* is a timeout for job closing, ``0.1`` by default.
      If job's closing time takes more than timeout a
      message is logged by :meth:`Scheduler.call_exception_handler`.
 
-   * *limit* is a for jobs spawned by scheduler, ``100`` by
+   * *limit* is a limit for jobs spawned by scheduler, ``100`` by
      default.
 
    * *pending_limit* is a limit for amount of jobs awaiting starting,
@@ -44,29 +54,10 @@ Instantiation
      for everybody, user should pass a value suitable for his
      environment anyway.
 
-
-Scheduler
----------
-
-.. class:: Scheduler
-
-   A container for managed jobs.
-
-   Jobs are created by :meth:`spawn()`.
-
-   :meth:`close` should be used for finishing all scheduled jobs.
-
-   The class implements :class:`collections.abc.Collection` contract,
-   jobs could be iterated etc.: ``len(scheduler)``, ``for job in
-   scheduler``, ``job in scheduler`` operations are supported.
-
-   User should never instantiate the class but call
-   :func:`create_scheduler` async function.
-
    .. attribute:: limit
 
       Concurrency limit (``100`` by default) or ``None`` if the limit
-      is disabled. See :func:`create_scheduler` for setting the attribute.
+      is disabled.
 
    .. attribute:: pending_limit
 
@@ -135,7 +126,7 @@ Scheduler
       By default calls
       :meth:`asyncio.AbstractEventLoop.call_exception_handler`, the
       behavior could be overridden by passing *exception_handler*
-      parameter into :func:`create_scheduler`.
+      parameter into :class:`Scheduler`.
 
       *context* is a :class:`dict` with the following keys:
 
@@ -213,8 +204,7 @@ jobs.
    closing it on web server shutdown.
 
    * *app* - :class:`aiohttp.web.Application` instance.
-   * *kwargs* - additional named parameters passed to
-     :func:`aiojobs.create_scheduler`.
+   * *kwargs* - additional named parameters passed to :class:`aiojobs.Scheduler`.
 
 .. cofunction:: spawn(request, coro)
 
diff --git a/docs/index.rst b/docs/index.rst
index f7b7230..1660ae1 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -28,7 +28,7 @@ Usage example
        await asyncio.sleep(timeout)
 
    async def main():
-       scheduler = await aiojobs.create_scheduler()
+       scheduler = aiojobs.Scheduler()
        for i in range(100):
            # spawn jobs
            await scheduler.spawn(coro(i/10))
diff --git a/docs/intro.rst b/docs/intro.rst
index 60abc00..20cd0d2 100644
--- a/docs/intro.rst
+++ b/docs/intro.rst
@@ -55,7 +55,7 @@ Scheduler's jobs could be enumerated and closed.
 
 There is simple usage example::
 
-   scheduler = await aiojobs.create_scheduler()
+   scheduler = aiojobs.Scheduler()
 
    job = await scheduler.spawn(f())
 
@@ -88,9 +88,8 @@ It prevents a program over-flooding by running a billion of jobs at
 the same time.
 
 The limit could be relaxed by passing *limit* parameter into
-:func:`create_scheduler`: ``await
-aiojobs.create_scheduler(limit=100000)`` or even disabled by
-``limit=None``.
+:class:`Scheduler`: ``aiojobs.Scheduler(limit=100000)`` or even
+disabled by ``limit=None``.
 
 Graceful Shutdown
 -----------------
@@ -103,8 +102,8 @@ The call has a timeout for waiting for close:
 If spawned job's closing time takes more than timeout a message is logged by
 :meth:`Scheduler.call_exception_handler`.
 
-Close timeout could be overridden by :func:`create_scheduler`: ``await
-aiojobs.create_scheduler(close_timeout=10)``
+Close timeout could be overridden in :class:`Scheduler`:
+``aiojobs.Scheduler(close_timeout=10)``
 
 
 Introspection
diff --git a/docs/quickstart.rst b/docs/quickstart.rst
index 57c4767..faeeefc 100644
--- a/docs/quickstart.rst
+++ b/docs/quickstart.rst
@@ -27,7 +27,7 @@ Instantiate a scheduler::
 
    import aiojobs
 
-   scheduler = await aiojobs.create_scheduler()
+   scheduler = aiojobs.Scheduler()
 
 Spawn a new job::
 
@@ -47,7 +47,7 @@ Let's collect it altogether into very small but still functional example::
        await asyncio.sleep(timeout)
 
    async def main():
-       scheduler = await aiojobs.create_scheduler()
+       scheduler = aiojobs.Scheduler()
        for i in range(100):
            # spawn jobs
            await scheduler.spawn(coro(i/10))


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index ed44d9a..b0991bf 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -3,8 +3,7 @@ from typing import Any, AsyncIterator, Awaitable, Callable, Dict
 
 import pytest
 
-from aiojobs import create_scheduler
-from aiojobs._scheduler import Scheduler
+from aiojobs import Scheduler
 
 PARAMS: Dict[str, Any] = dict(
     close_timeout=1.0, limit=100, pending_limit=0, exception_handler=None
@@ -23,7 +22,7 @@ async def make_scheduler() -> AsyncIterator[Callable[..., Awaitable[Scheduler]]]
     schedulers = []
 
     async def maker(**kwargs: Any) -> Scheduler:
-        ret = await create_scheduler(**kwargs)
+        ret = Scheduler(**kwargs)
         schedulers.append(ret)
         return ret
 
diff --git a/tests/test_job.py b/tests/test_job.py
index 8cb19e6..73d4a10 100644
--- a/tests/test_job.py
+++ b/tests/test_job.py
@@ -112,7 +112,7 @@ async def test_job_wait_exception(make_scheduler: _MakeScheduler) -> None:
     scheduler = await make_scheduler(exception_handler=handler)
     exc = RuntimeError()
 
-    async def coro() -> None:
+    async def coro() -> NoReturn:
         raise exc
 
     job = await scheduler.spawn(coro())
@@ -275,3 +275,15 @@ async def test_job_await_explicit_close(scheduler: Scheduler) -> None:
     assert job._task.done()  # type: ignore[unreachable]
     with pytest.raises(asyncio.CancelledError):
         await job.wait()
+
+
+async def test_exception_handler_called_once(make_scheduler: _MakeScheduler) -> None:
+    handler = mock.Mock()
+    scheduler = await make_scheduler(exception_handler=handler)
+
+    async def coro() -> NoReturn:
+        raise Exception()
+
+    await scheduler.spawn(coro())
+    await scheduler.close()
+    handler.assert_called_once()