Add pool creation functions

Added :func:`_sa.create_pool_from_url` and
:func:`_asyncio.create_async_pool_from_url` to create
a :class:`_pool.Pool` instance from an input url passed as string
or :class:`_sa.URL`.

Fixes: #9613
Change-Id: Icd8aa3f2849e6fd1bc5341114f3ef8d216a2c543

12 files changed, 194 insertions, 13 deletions

diff --git a/doc/build/changelog/changelog_20.rst b/doc/build/changelog/changelog_20.rst
index 1319c8293..b26e071eb 100644
--- a/doc/build/changelog/changelog_20.rst
+++ b/doc/build/changelog/changelog_20.rst
@@ -651,7 +651,7 @@
         use is not supported and will be removed in a future release.
 
     .. change::
-        :tags: orm, use_case
+        :tags: orm, usecase
         :tickets: 9297
 
         To accommodate a change in column ordering used by ORM Declarative in
diff --git a/doc/build/changelog/unreleased_20/9613.rst b/doc/build/changelog/unreleased_20/9613.rst
new file mode 100644
index 000000000..1c44eb1ec
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/9613.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: usecase, pool
+    :tickets: 9625
+
+    Added :func:`_sa.create_pool_from_url` and
+    :func:`_asyncio.create_async_pool_from_url` to create
+    a :class:`_pool.Pool` instance from an input url passed as string
+    or :class:`_sa.URL`.
diff --git a/doc/build/core/engines.rst b/doc/build/core/engines.rst
index a41f8b984..696f058e6 100644
--- a/doc/build/core/engines.rst
+++ b/doc/build/core/engines.rst
@@ -269,6 +269,7 @@ Engine Creation API
 
 .. autofunction:: sqlalchemy.engine.make_url
 
+.. autofunction:: sqlalchemy.create_pool_from_url
 
 .. autoclass:: sqlalchemy.engine.URL
     :members:
diff --git a/doc/build/core/pooling.rst b/doc/build/core/pooling.rst
index b47549e88..8e7f1592d 100644
--- a/doc/build/core/pooling.rst
+++ b/doc/build/core/pooling.rst
@@ -683,6 +683,35 @@ ORM :class:`_orm.Session` object that's begun a transaction and references
 active :class:`_orm.Connection` instances; again prefer to create new
 :class:`_orm.Session` objects in new processes.
 
+Using a pool instance directly
+------------------------------
+
+A pool implementation can be used directly without an engine. This could be used
+in applications that just whish to use the pool behavior without all other
+SQLAlchemy features.
+In the example below the default pool for the ``MySQLdb`` dialect is obtained using
+:func:`_sa.create_pool_from_url`::
+
+    from sqlalchemy import create_pool_from_url
+
+    my_pool = create_pool_from_url(
+        "mysql+mysqldb://", max_overflow=5, pool_size=5, pre_ping=True
+    )
+
+    con = my_pool.connect()
+    # use the connection
+    ...
+    # then close it
+    con.close()
+
+If the type of pool to create is not specified, the default one for the dialect
+will be used. To specify it directly the ``poolclass`` argument can be used,
+like in the following example::
+
+    from sqlalchemy import create_pool_from_url
+    from sqlalchemy import NullPool
+
+    my_pool = create_pool_from_url("mysql+mysqldb://", poolclass=NullPool)
 
 API Documentation - Available Pool Implementations
 --------------------------------------------------
diff --git a/doc/build/orm/extensions/asyncio.rst b/doc/build/orm/extensions/asyncio.rst
index 59989ad4e..c57f1199c 100644
--- a/doc/build/orm/extensions/asyncio.rst
+++ b/doc/build/orm/extensions/asyncio.rst
@@ -970,6 +970,8 @@ Engine API Documentation
 
 .. autofunction:: async_engine_from_config
 
+.. autofunction:: create_async_pool_from_url
+
 .. autoclass:: AsyncEngine
    :members:
 
diff --git a/lib/sqlalchemy/__init__.py b/lib/sqlalchemy/__init__.py
index 935c160cf..86c5afd8f 100644
--- a/lib/sqlalchemy/__init__.py
+++ b/lib/sqlalchemy/__init__.py
@@ -18,6 +18,7 @@ from .engine import Compiled as Compiled
 from .engine import Connection as Connection
 from .engine import create_engine as create_engine
 from .engine import create_mock_engine as create_mock_engine
+from .engine import create_pool_from_url as create_pool_from_url
 from .engine import CreateEnginePlugin as CreateEnginePlugin
 from .engine import CursorResult as CursorResult
 from .engine import Dialect as Dialect
diff --git a/lib/sqlalchemy/engine/__init__.py b/lib/sqlalchemy/engine/__init__.py
index 09ff9a787..843f97025 100644
--- a/lib/sqlalchemy/engine/__init__.py
+++ b/lib/sqlalchemy/engine/__init__.py
@@ -24,6 +24,7 @@ from .base import RootTransaction as RootTransaction
 from .base import Transaction as Transaction
 from .base import TwoPhaseTransaction as TwoPhaseTransaction
 from .create import create_engine as create_engine
+from .create import create_pool_from_url as create_pool_from_url
 from .create import engine_from_config as engine_from_config
 from .cursor import CursorResult as CursorResult
 from .cursor import ResultProxy as ResultProxy
diff --git a/lib/sqlalchemy/engine/create.py b/lib/sqlalchemy/engine/create.py
index c491240ea..bddf51fb6 100644
--- a/lib/sqlalchemy/engine/create.py
+++ b/lib/sqlalchemy/engine/create.py
@@ -29,6 +29,7 @@ from .. import util
 from ..pool import _AdhocProxiedConnection
 from ..pool import ConnectionPoolEntry
 from ..sql import compiler
+from ..util import immutabledict
 
 if typing.TYPE_CHECKING:
     from .base import Engine
@@ -644,18 +645,8 @@ def create_engine(url: Union[str, _url.URL], **kwargs: Any) -> Engine:
 
         # consume pool arguments from kwargs, translating a few of
         # the arguments
-        translate = {
-            "logging_name": "pool_logging_name",
-            "echo": "echo_pool",
-            "timeout": "pool_timeout",
-            "recycle": "pool_recycle",
-            "events": "pool_events",
-            "reset_on_return": "pool_reset_on_return",
-            "pre_ping": "pool_pre_ping",
-            "use_lifo": "pool_use_lifo",
-        }
         for k in util.get_cls_kwargs(poolclass):
-            tk = translate.get(k, k)
+            tk = _pool_translate_kwargs.get(k, k)
             if tk in kwargs:
                 pool_args[k] = pop_kwarg(tk)
 
@@ -811,3 +802,60 @@ def engine_from_config(
     options.update(kwargs)
     url = options.pop("url")
     return create_engine(url, **options)
+
+
+@overload
+def create_pool_from_url(
+    url: Union[str, URL],
+    *,
+    poolclass: Optional[Type[Pool]] = ...,
+    logging_name: str = ...,
+    pre_ping: bool = ...,
+    size: int = ...,
+    recycle: int = ...,
+    reset_on_return: Optional[_ResetStyleArgType] = ...,
+    timeout: float = ...,
+    use_lifo: bool = ...,
+    **kwargs: Any,
+) -> Pool:
+    ...
+
+
+@overload
+def create_pool_from_url(url: Union[str, URL], **kwargs: Any) -> Pool:
+    ...
+
+
+def create_pool_from_url(url: Union[str, URL], **kwargs: Any) -> Pool:
+    """Create a pool instance from the given url.
+
+    If ``poolclass`` is not provided the pool class used
+    is selected using the dialect specified in the URL.
+
+    The arguments passed to :func:`_sa.create_pool_from_url` are
+    identical to the pool argument passed to the :func:`_sa.create_engine`
+    function.
+
+    .. versionadded:: 2.0.10
+    """
+
+    for key in _pool_translate_kwargs:
+        if key in kwargs:
+            kwargs[_pool_translate_kwargs[key]] = kwargs.pop(key)
+
+    engine = create_engine(url, **kwargs, _initialize=False)
+    return engine.pool
+
+
+_pool_translate_kwargs = immutabledict(
+    {
+        "logging_name": "pool_logging_name",
+        "echo": "echo_pool",
+        "timeout": "pool_timeout",
+        "recycle": "pool_recycle",
+        "events": "pool_events",  # deprecated
+        "reset_on_return": "pool_reset_on_return",
+        "pre_ping": "pool_pre_ping",
+        "use_lifo": "pool_use_lifo",
+    }
+)
diff --git a/lib/sqlalchemy/ext/asyncio/__init__.py b/lib/sqlalchemy/ext/asyncio/__init__.py
index 943734928..7195e1f07 100644
--- a/lib/sqlalchemy/ext/asyncio/__init__.py
+++ b/lib/sqlalchemy/ext/asyncio/__init__.py
@@ -10,6 +10,7 @@ from .engine import AsyncConnection as AsyncConnection
 from .engine import AsyncEngine as AsyncEngine
 from .engine import AsyncTransaction as AsyncTransaction
 from .engine import create_async_engine as create_async_engine
+from .engine import create_async_pool_from_url as create_async_pool_from_url
 from .result import AsyncMappingResult as AsyncMappingResult
 from .result import AsyncResult as AsyncResult
 from .result import AsyncScalarResult as AsyncScalarResult
diff --git a/lib/sqlalchemy/ext/asyncio/engine.py b/lib/sqlalchemy/ext/asyncio/engine.py
index 325c58bda..440cf834d 100644
--- a/lib/sqlalchemy/ext/asyncio/engine.py
+++ b/lib/sqlalchemy/ext/asyncio/engine.py
@@ -35,6 +35,7 @@ from ... import inspection
 from ... import util
 from ...engine import Connection
 from ...engine import create_engine as _create_engine
+from ...engine import create_pool_from_url as _create_pool_from_url
 from ...engine import Engine
 from ...engine.base import NestedTransaction
 from ...engine.base import Transaction
@@ -80,7 +81,6 @@ def create_async_engine(url: Union[str, URL], **kw: Any) -> AsyncEngine:
             "use the connection.stream() method for an async "
             "streaming result set"
         )
-    kw["future"] = True
     kw["_is_async"] = True
     sync_engine = _create_engine(url, **kw)
     return AsyncEngine(sync_engine)
@@ -111,6 +111,21 @@ def async_engine_from_config(
     return create_async_engine(url, **options)
 
 
+def create_async_pool_from_url(url: Union[str, URL], **kwargs: Any) -> Pool:
+    """Create a new async engine instance.
+
+    Arguments passed to :func:`_asyncio.create_async_pool_from_url` are mostly
+    identical to those passed to the :func:`_sa.create_pool_from_url` function.
+    The specified dialect must be an asyncio-compatible dialect
+    such as :ref:`dialect-postgresql-asyncpg`.
+
+    .. versionadded:: 2.0.10
+
+    """
+    kwargs["_is_async"] = True
+    return _create_pool_from_url(url, **kwargs)
+
+
 class AsyncConnectable:
     __slots__ = "_slots_dispatch", "__weakref__"
 
diff --git a/test/engine/test_parseconnect.py b/test/engine/test_parseconnect.py
index f571b4bab..471201666 100644
--- a/test/engine/test_parseconnect.py
+++ b/test/engine/test_parseconnect.py
@@ -2,9 +2,11 @@ import copy
 from unittest.mock import call
 from unittest.mock import MagicMock
 from unittest.mock import Mock
+from unittest.mock import patch
 
 import sqlalchemy as tsa
 from sqlalchemy import create_engine
+from sqlalchemy import create_pool_from_url
 from sqlalchemy import engine_from_config
 from sqlalchemy import exc
 from sqlalchemy import pool
@@ -13,9 +15,11 @@ from sqlalchemy.dialects import plugins
 from sqlalchemy.dialects import registry
 from sqlalchemy.engine.default import DefaultDialect
 import sqlalchemy.engine.url as url
+from sqlalchemy.pool.impl import NullPool
 from sqlalchemy.testing import assert_raises
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import eq_
+from sqlalchemy.testing import fixture
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
 from sqlalchemy.testing import is_false
@@ -869,6 +873,56 @@ class CreateEngineTest(fixtures.TestBase):
         ne_(successes, 0, "No default drivers found.")
 
 
+class CreatePoolTest(fixtures.TestBase):
+    @fixture
+    def mock_create(self):
+        with patch(
+            "sqlalchemy.engine.create.create_engine",
+        ) as p:
+            yield p
+
+    def test_url_only(self, mock_create):
+        create_pool_from_url("sqlite://")
+        mock_create.assert_called_once_with("sqlite://", _initialize=False)
+
+    def test_pool_args(self, mock_create):
+        create_pool_from_url(
+            "sqlite://",
+            logging_name="foo",
+            echo=True,
+            timeout=42,
+            recycle=22,
+            reset_on_return=True,
+            pre_ping=True,
+            use_lifo=True,
+            foo=99,
+        )
+        mock_create.assert_called_once_with(
+            "sqlite://",
+            pool_logging_name="foo",
+            echo_pool=True,
+            pool_timeout=42,
+            pool_recycle=22,
+            pool_reset_on_return=True,
+            pool_pre_ping=True,
+            pool_use_lifo=True,
+            foo=99,
+            _initialize=False,
+        )
+
+    def test_pool_creation(self):
+        pp = create_pool_from_url("sqlite://")
+        engine_pool = create_engine("sqlite://").pool
+        eq_(pp.__class__, engine_pool.__class__)
+        pp = create_pool_from_url("sqlite://", pre_ping=True)
+        is_true(pp._pre_ping)
+        is_false(isinstance(pp, NullPool))
+
+    def test_pool_creation_custom_class(self):
+        pp = create_pool_from_url("sqlite://", poolclass=NullPool)
+        is_true(isinstance(pp, NullPool))
+
+
 class TestRegNewDBAPI(fixtures.TestBase):
     def test_register_base(self):
         registry.register("mockdialect", __name__, "MockDialect")
diff --git a/test/ext/asyncio/test_engine_py3k.py b/test/ext/asyncio/test_engine_py3k.py
index 9511fed74..786f841ee 100644
--- a/test/ext/asyncio/test_engine_py3k.py
+++ b/test/ext/asyncio/test_engine_py3k.py
@@ -1,5 +1,6 @@
 import asyncio
 import inspect as stdlib_inspect
+from unittest.mock import patch
 
 from sqlalchemy import Column
 from sqlalchemy import create_engine
@@ -18,6 +19,7 @@ from sqlalchemy import union_all
 from sqlalchemy.engine import cursor as _cursor
 from sqlalchemy.ext.asyncio import async_engine_from_config
 from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.ext.asyncio import create_async_pool_from_url
 from sqlalchemy.ext.asyncio import engine as _async_engine
 from sqlalchemy.ext.asyncio import exc as async_exc
 from sqlalchemy.ext.asyncio import exc as asyncio_exc
@@ -707,6 +709,25 @@ class AsyncEngineTest(EngineFixture):
         assert engine.dialect.is_async is True
 
 
+class AsyncCreatePoolTest(fixtures.TestBase):
+    @config.fixture
+    def mock_create(self):
+        with patch(
+            "sqlalchemy.ext.asyncio.engine._create_pool_from_url",
+        ) as p:
+            yield p
+
+    def test_url_only(self, mock_create):
+        create_async_pool_from_url("sqlite://")
+        mock_create.assert_called_once_with("sqlite://", _is_async=True)
+
+    def test_pool_args(self, mock_create):
+        create_async_pool_from_url("sqlite://", foo=99, echo=True)
+        mock_create.assert_called_once_with(
+            "sqlite://", foo=99, echo=True, _is_async=True
+        )
+
+
 class AsyncEventTest(EngineFixture):
     """The engine events all run in their normal synchronous context.