ExecutionContext
is a class within the sqlalchemy.engine.interfaces
module of the SQLAlchemy project.
databases (project homepage and PyPI page provides asyncio support with an SQLALchemy Core interface for common relational databases such as MySQL, PostgreSQL and SQLite. This is handy for integrating with asynchronous I/O web frameworks like Sanic. The project is open sourced under the BSD 3-Clause "New" or "Revised" License.
databases / databases / backends / sqlite.py
# sqlite.py
import logging
import typing
import uuid
import aiosqlite
from sqlalchemy.dialects.sqlite import pysqlite
from sqlalchemy.engine.interfaces import Dialect, ExecutionContext
from sqlalchemy.engine.result import ResultMetaData, RowProxy
from sqlalchemy.sql import ClauseElement
from sqlalchemy.types import TypeEngine
from databases.core import LOG_EXTRA, DatabaseURL
from databases.interfaces import ConnectionBackend, DatabaseBackend, TransactionBackend
logger = logging.getLogger("databases")
class SQLiteBackend(DatabaseBackend):
def __init__(
self, database_url: typing.Union[DatabaseURL, str], **options: typing.Any
) -> None:
self._database_url = DatabaseURL(database_url)
self._options = options
self._dialect = pysqlite.dialect(paramstyle="qmark")
self._dialect.supports_native_decimal = False
self._pool = SQLitePool(self._database_url, **self._options)
async def connect(self) -> None:
pass
async def disconnect(self) -> None:
## ... source file continues with no further ExecutionContext examples...