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

1 files changed, 59 insertions, 11 deletions

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",
+    }
+)