Create a new AsyncEngine instance using a configuration dictionary. This function is analogous to the :func:`_sa.engine_from_config` function in SQLAlchemy Core, except that the requested dialect must be an asyncio-compatible dialect such as :ref:`dialect-postgresql-asyncpg`. The ar
(
configuration: Dict[str, Any], prefix: str = "sqlalchemy.", **kwargs: Any
)
| 122 | |
| 123 | |
| 124 | def async_engine_from_config( |
| 125 | configuration: Dict[str, Any], prefix: str = "sqlalchemy.", **kwargs: Any |
| 126 | ) -> AsyncEngine: |
| 127 | """Create a new AsyncEngine instance using a configuration dictionary. |
| 128 | |
| 129 | This function is analogous to the :func:`_sa.engine_from_config` function |
| 130 | in SQLAlchemy Core, except that the requested dialect must be an |
| 131 | asyncio-compatible dialect such as :ref:`dialect-postgresql-asyncpg`. |
| 132 | The argument signature of the function is identical to that |
| 133 | of :func:`_sa.engine_from_config`. |
| 134 | |
| 135 | .. versionadded:: 1.4.29 |
| 136 | |
| 137 | """ |
| 138 | options = { |
| 139 | key[len(prefix) :]: value |
| 140 | for key, value in configuration.items() |
| 141 | if key.startswith(prefix) |
| 142 | } |
| 143 | options["_coerce_config"] = True |
| 144 | options.update(kwargs) |
| 145 | url = options.pop("url") |
| 146 | return create_async_engine(url, **options) |
| 147 | |
| 148 | |
| 149 | def create_async_pool_from_url(url: Union[str, URL], **kwargs: Any) -> Pool: |