diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 2edc747..70c9ffb 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -15,10 +15,8 @@ jobs: env: QUERY_DEFAULT_USER: databend QUERY_DEFAULT_PASSWORD: databend - MINIO_ENABLED: true ports: - 8000:8000 - - 9000:9000 steps: - name: Checkout uses: actions/checkout@v4 @@ -44,3 +42,12 @@ jobs: TEST_DATABEND_DSN: "databend://databend:databend@localhost:8000/default?sslmode=disable" run: | pipenv run pytest -s . + + - name: SQLAlchemy Test Suite (asyncio) + run: | + pipenv run pytest -s . --db async + + - name: Tests on SQLAlchemy 1.4 + run: | + pipenv run pip install "sqlalchemy<2" + pipenv run pytest -s . diff --git a/Pipfile b/Pipfile index bbd4705..a34076a 100644 --- a/Pipfile +++ b/Pipfile @@ -4,7 +4,7 @@ verify_ssl = true name = "pypi" [packages] -databend-sqlalchemy = {file = ".", editable = true} +databend-sqlalchemy = {file = ".", editable = true, extras = ["asyncio"]} [dev-packages] black = "*" diff --git a/README.rst b/README.rst index d6d960d..15db0bc 100644 --- a/README.rst +++ b/README.rst @@ -33,6 +33,49 @@ The DSN format is similar to that of regular Postgres:: print(row) +Asyncio Support +--------------- + +With SQLAlchemy 2.0.28 or later, the ``databend+async`` dialect runs on the asyncio flavour of +the databend driver. Install it with the ``asyncio`` extra, which brings in greenlet:: + + pip install "databend-sqlalchemy[asyncio]" + +Create the engine with ``create_async_engine``, which also picks this dialect for a plain +``databend://`` URL:: + + from sqlalchemy import text + from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine + + engine = create_async_engine( + f"databend+async://{username}:{password}@{host_port_name}/{database_name}?sslmode=disable" + ) + async with engine.connect() as connection: + result = await connection.execute(text("SELECT 1")) + assert result.scalar() == 1 + + # rows are fetched from Databend as they are consumed, rather than buffered + result = await connection.stream(text("SELECT number FROM numbers(1000000)")) + async for partition in result.partitions(10000): + ... + + async with AsyncSession(engine) as session: + ... + +Each pooled connection holds one Databend session, so session settings and temporary tables +persist across its statements. + +The async connector can also be used directly:: + + from databend_sqlalchemy import async_connector + + connection = await async_connector.connect('databend://root:@localhost:8000?sslmode=disable') + cursor = connection.cursor() + await cursor.execute('SELECT * FROM test') + async for row in cursor: + print(row) + + Merge Command Support --------------------- @@ -63,7 +106,7 @@ The Merge command can be used as below:: Copy Into Command Support ---------------------- +------------------------- Databend SQLAlchemy supports copy into operations through it's CopyIntoTable and CopyIntoLocation methods See [CopyIntoLocation](https://docs.databend.com/sql/sql-commands/dml/dml-copy-into-location) or [CopyIntoTable](https://docs.databend.com/sql/sql-commands/dml/dml-copy-into-table) for full documentation. @@ -222,5 +265,8 @@ The table options can be used as below:: Compatibility --------------- +- databend-sqlalchemy needs Python 3.10 or later and SQLAlchemy 1.4 or later. The asyncio dialect needs + SQLAlchemy 2.0.28 or later, as do result row counts for INSERT, UPDATE, DELETE and COPY INTO, which + earlier releases report as -1. - If databend version >= v0.9.0 or later, you need to use databend-sqlalchemy version >= v0.1.0. - The databend-sqlalchemy use [databend-py](https://github.com/databendlabs/databend-py) as internal driver when version < v0.4.0, but when version >= v0.4.0 it use [databend driver python binding](https://github.com/databendlabs/bendsql/blob/main/bindings/python/README.md) as internal driver. The only difference between the two is that the connection parameters provided in the DSN are different. When using the corresponding version, you should refer to the connection parameters provided by the corresponding Driver. diff --git a/databend_sqlalchemy/async_connector.py b/databend_sqlalchemy/async_connector.py new file mode 100644 index 0000000..aa2520b --- /dev/null +++ b/databend_sqlalchemy/async_connector.py @@ -0,0 +1,204 @@ +#!/usr/bin/env python +""" +Asyncio counterpart of :mod:`databend_sqlalchemy.connector`, built on the driver's +``AsyncDatabendClient``. + +The driver offers no async cursor, so :class:`AsyncCursor` supplies the :pep:`249` +cursor surface as coroutines, rendering parameters exactly as the blocking cursor does. +Unlike the blocking connector, which opens a driver connection per cursor, every cursor +of an :class:`AsyncConnection` shares one driver connection, and so one Databend +session. +""" +import asyncio +from collections import deque +from collections.abc import Sequence +from itertools import islice +from types import TracebackType + +from databend_driver import ( + AsyncDatabendClient, + AsyncDatabendConnection, + Row, + RowIterator, +) + +from databend_sqlalchemy.connector import ( # noqa: F401 - PEP 249 module globals + Binary, + OperationRenderer, + Parameters, + apilevel, + paramstyle, + threadsafety, +) +from databend_sqlalchemy.errors import Error, NotSupportedError + +ColumnDescription = tuple[str, str, None, None, None, None, None] + + +async def connect( + dsn: str = "databend://root:@localhost:8000/?sslmode=disable", +) -> "AsyncConnection": + try: + driver_connection = await AsyncDatabendClient(dsn).get_conn() + except Exception as e: + raise Error(str(e)) from e + return AsyncConnection(driver_connection) + + +class AsyncConnection: + def __init__(self, driver_connection: AsyncDatabendConnection) -> None: + self.driver_connection = driver_connection + + async def close(self) -> None: + """Close the Databend session, where the driver is new enough to offer that.""" + close_session = getattr(self.driver_connection, "close", None) + if close_session is not None: + await close_session() + + async def commit(self) -> None: + pass + + def cursor(self) -> "AsyncCursor": + return AsyncCursor(self.driver_connection) + + async def rollback(self) -> None: + raise NotSupportedError("Transactions are not supported") # pragma: no cover + + +class AsyncCursor(OperationRenderer): + """Asyncio counterpart of :class:`databend_sqlalchemy.connector.Cursor`. + + As with the blocking driver cursor, every statement runs through ``query_iter`` and + its first row is read straight away, so a statement without a result set has + finished by the time :meth:`execute` returns. + """ + + arraysize = 1 + + def __init__(self, driver_connection: AsyncDatabendConnection) -> None: + self.driver_connection = driver_connection + self.description: list[ColumnDescription] | None = None + self._rows: RowIterator | None = None + self._read_ahead: deque[Row] = deque() + + @property + def rowcount(self) -> int: + """By default, return -1 to indicate that this is not supported.""" + return -1 + + async def __aenter__(self) -> "AsyncCursor": + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + await self.close() + + async def close(self) -> None: + """Discard any unread rows, leaving the shared session open.""" + self._discard_results() + + async def execute( + self, operation: str, parameters: Parameters | None = None + ) -> None: + """Prepare and execute a database operation (query or command). + + Blank operations, such as the empty DDL compiled for unsupported index + statements, are skipped just as the blocking cursor skips them. + """ + if not operation: + return + try: + await self._run_query(self.render_operation(operation, parameters)) + except Exception as e: + raise Error(str(e)) from e + + async def executemany( + self, operation: str, seq_of_parameters: Sequence[Parameters] + ) -> None: + """Run ``operation`` once per parameter set, sending ``INSERT ... VALUES`` as a + single multi-row statement. Only the final result set is retained.""" + try: + query = self.render_multi_row_insert(operation, seq_of_parameters) + if query is not None: + return await self._run_query(query) + except Exception as e: + raise Error(str(e)) from e + for parameters in seq_of_parameters: + await self.execute(operation, parameters) + + async def fetchone(self) -> tuple | None: + row = await self._next_row() + return None if row is None else row.values() + + async def fetchmany(self, size: int | None = None) -> list[tuple]: + return await self._fetch_rows(self.arraysize if size is None else size) + + async def fetchall(self) -> list[tuple]: + return await self._fetch_rows() + + def __aiter__(self) -> "AsyncCursor": + return self + + async def __anext__(self) -> tuple: + row = await self.fetchone() + if row is None: + raise StopAsyncIteration + return row + + def _discard_results(self) -> None: + self.description = None + self._rows = None + self._read_ahead.clear() + + async def _run_query(self, query: str) -> None: + self._discard_results() + rows = await self.driver_connection.query_iter(query) + self.description = [ + (field.name, field.data_type, None, None, None, None, None) + for field in rows.schema().fields() + ] + self._rows = rows + if (first_row := await anext(rows, None)) is not None: + self._read_ahead.append(first_row) + + async def _next_row(self) -> Row | None: + if self._read_ahead: + return self._read_ahead.popleft() + if self._rows is None: + return None + try: + return await anext(self._rows, None) + except Exception as e: + raise Error(str(e)) from e + + async def _fetch_rows(self, limit: int | None = None) -> list[tuple]: + """Return up to ``limit`` rows, or every remaining row when ``limit`` is + ``None``. + + Awaiting the driver's iterator costs a trip through its runtime for every row, + which is around a hundred times slower than iterating it directly. Rows beyond + the read-ahead are therefore drained with the iterator's blocking ``__next__`` + in a worker thread, one trip per batch. The iterator is released once + exhausted. + """ + rows = [] + while self._read_ahead and (limit is None or len(rows) < limit): + rows.append(self._read_ahead.popleft().values()) + wanted = None if limit is None else limit - len(rows) + if self._rows is None or wanted == 0: + return rows + batch = await asyncio.to_thread(self._drain_rows, self._rows, wanted) + if wanted is None or len(batch) < wanted: + self._rows = None + return rows + batch + + @staticmethod + def _drain_rows(rows: RowIterator, limit: int | None) -> list[tuple]: + try: + return [row.values() for row in islice(rows, limit)] + except Exception as e: + raise Error(str(e)) from e diff --git a/databend_sqlalchemy/async_dialect.py b/databend_sqlalchemy/async_dialect.py new file mode 100644 index 0000000..0ad6852 --- /dev/null +++ b/databend_sqlalchemy/async_dialect.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python +""" +Databend Asyncio Support +------------------------ + +The ``databend+async`` dialect runs on the asyncio flavour of ``databend-driver`` +through :mod:`databend_sqlalchemy.async_connector`, and needs SQLAlchemy 2.0.28 or +later, the first release to report the row counts Databend returns for inserts:: + + from sqlalchemy.ext.asyncio import create_async_engine + + engine = create_async_engine("databend+async://user:password@host:8000/database") + +``create_async_engine`` given a plain ``databend://`` URL selects this dialect as well. + +Results are buffered in full when a statement executes, unless streamed with +``AsyncConnection.stream()`` or the ``stream_results`` execution option, which fetch +rows from Databend as they are consumed. +""" +import re +from collections import deque +from collections.abc import Sequence +from types import ModuleType +from typing import Any + +import sqlalchemy +from sqlalchemy import pool +from sqlalchemy.engine.url import URL +from sqlalchemy.util.concurrency import await_only + +from databend_sqlalchemy import async_connector +from databend_sqlalchemy.connector import Parameters +from databend_sqlalchemy.databend_dialect import ( + DatabendDialect, + DatabendExecutionContext, +) + +ASYNCIO_REQUIREMENT = ( + "databend+async needs SQLAlchemy 2.0.28 or later: " + 'pip install "databend-sqlalchemy[asyncio]"' +) + +try: + from sqlalchemy.connectors.asyncio import ( + AsyncAdapt_dbapi_connection, + AsyncAdapt_dbapi_cursor, + AsyncAdapt_dbapi_ss_cursor, + ) +except ImportError as missing_adapter: + raise ImportError(ASYNCIO_REQUIREMENT) from missing_adapter + + +def installed_sqlalchemy_release() -> tuple[int, ...]: + """Major, minor and patch number of the installed SQLAlchemy, ignoring any + pre-release tag.""" + release_numbers = re.findall(r"\d+", sqlalchemy.__version__)[:3] + return tuple(int(number) for number in release_numbers) + + +if installed_sqlalchemy_release() < (2, 0, 28): + raise ImportError(ASYNCIO_REQUIREMENT) + + +class AsyncAdaptedDatabendCursor(AsyncAdapt_dbapi_cursor): + """Buffers the rows an ``executemany`` leaves behind as well as those of + ``execute``, as :meth:`DatabendExecutionContext.post_exec` reads the inserted row + count back after both.""" + + __slots__ = () + + async def _executemany_async( + self, operation: str, seq_of_parameters: Sequence[Parameters] + ) -> None: + async with self._adapt_connection._execute_mutex: + await self._cursor.executemany(operation, seq_of_parameters) + if self._cursor.description: + self._rows = deque(await self._cursor.fetchall()) + + +class AsyncAdaptedDatabendConnection(AsyncAdapt_dbapi_connection): + __slots__ = () + + _cursor_cls = AsyncAdaptedDatabendCursor + + +class AsyncAdaptedDatabendDBAPI: + """Presents :mod:`databend_sqlalchemy.async_connector` to SQLAlchemy as a + synchronous DBAPI module, each await being bridged through the calling greenlet.""" + + def __init__(self, connector: ModuleType) -> None: + self.connector = connector + self.apilevel = connector.apilevel + self.threadsafety = connector.threadsafety + self.paramstyle = connector.paramstyle + self.Binary = connector.Binary + self.Error = connector.Error + self.NotSupportedError = connector.NotSupportedError + + def connect(self, *args: Any, **kwargs: Any) -> AsyncAdaptedDatabendConnection: + creator_fn = kwargs.pop("async_creator_fn", self.connector.connect) + return AsyncAdaptedDatabendConnection( + self, await_only(creator_fn(*args, **kwargs)) + ) + + +class AsyncDatabendExecutionContext(DatabendExecutionContext): + def create_server_side_cursor(self) -> AsyncAdapt_dbapi_ss_cursor: + return self._dbapi_connection.cursor(server_side=True) + + +class AsyncDatabendDialect(DatabendDialect): + driver = "async" + is_async = True + supports_statement_cache = False + + execution_ctx_cls = AsyncDatabendExecutionContext + + @classmethod + def import_dbapi(cls) -> AsyncAdaptedDatabendDBAPI: + return AsyncAdaptedDatabendDBAPI(async_connector) + + @classmethod + def get_pool_class(cls, url: URL) -> type[pool.Pool]: + return pool.AsyncAdaptedQueuePool + + def get_driver_connection( + self, connection: AsyncAdaptedDatabendConnection + ) -> async_connector.AsyncConnection: + return connection._connection + + +dialect = AsyncDatabendDialect diff --git a/databend_sqlalchemy/connector.py b/databend_sqlalchemy/connector.py index 28ad167..f2bb91e 100644 --- a/databend_sqlalchemy/connector.py +++ b/databend_sqlalchemy/connector.py @@ -6,7 +6,9 @@ import decimal import re import json +from collections.abc import Mapping, Sequence from datetime import datetime, date, time, timedelta +from typing import Any from databend_sqlalchemy.errors import Error, NotSupportedError from databend_driver import BlockingDatabendClient @@ -17,6 +19,15 @@ paramstyle = "pyformat" # Python extended format codes, e.g. ...WHERE name=%(name)s Binary = bytes # to satisfy dialect.dbapi.Binary +Parameters = Mapping[str, Any] | Sequence[Any] + +RE_INSERT_VALUES = re.compile( + r"\s*((?:INSERT|REPLACE)\s.+\sVALUES?\s*)" + + r"(\(\s*(?:%s|%\(.+\)s)\s*(?:,\s*(?:%s|%\(.+\)s)\s*)*\))" + + r"(\s*(?:ON DUPLICATE.*)?);?\s*\Z", + re.IGNORECASE | re.DOTALL, +) + class ParamEscaper: def escape_args(self, parameters): @@ -71,6 +82,38 @@ def escape_item(self, item): _escaper = ParamEscaper() +class OperationRenderer: + """Client-side parameter rendering shared by the blocking and asyncio cursors. + + Databend takes no bind parameters, so each operation is rendered to literal SQL + before it reaches the driver. + """ + + def mogrify(self, query: str, parameters: Parameters | None) -> str: + if parameters: + query = query % _escaper.escape_args(parameters) + return query + + def render_operation(self, operation: str, parameters: Parameters | None) -> str: + return self.mogrify(operation, parameters).replace("%%", "%") + + def render_multi_row_insert( + self, operation: str, seq_of_parameters: Sequence[Parameters] + ) -> str | None: + """Collapse an ``executemany`` of ``INSERT ... VALUES`` into a single multi-row + ``INSERT``, or return ``None`` when the operation is not of that shape.""" + m = RE_INSERT_VALUES.match(operation) + if not m: + return None + q_prefix = m.group(1) + q_values = m.group(2).rstrip() + values_list = [ + q_values % _escaper.escape_args(parameters) + for parameters in seq_of_parameters + ] + return "{} {};".format(q_prefix, ",".join(values_list)) + + # Patch ORM library @classmethod def create_ad_hoc_field(cls, db_type): @@ -124,7 +167,7 @@ def rollback(self): raise NotSupportedError("Transactions are not supported") # pragma: no cover -class Cursor: +class Cursor(OperationRenderer): """These objects represent a database cursor, which is used to manage the context of a fetch operation. @@ -168,12 +211,7 @@ def close(self): except Exception as e: raise Error(str(e)) from e - def mogrify(self, query, parameters): - if parameters: - query = query % _escaper.escape_args(parameters) - return query - - def execute(self, operation, parameters=None): + def execute(self, operation: str, parameters: Parameters | None = None) -> None: """Prepare and execute a database operation (query or command).""" # ToDo - Fix this, which is preventing the execution of blank DDL such as CREATE INDEX statements which aren't currently supported @@ -182,13 +220,13 @@ def execute(self, operation, parameters=None): return try: - query = self.mogrify(operation, parameters) - query = query.replace("%%", "%") - return self.inner.execute(query) + return self.inner.execute(self.render_operation(operation, parameters)) except Exception as e: raise Error(str(e)) from e - def executemany(self, operation, seq_of_parameters): + def executemany( + self, operation: str, seq_of_parameters: Sequence[Parameters] + ) -> None: """Prepare a database operation (query or command) and then execute it against all parameter sequences or mappings found in the sequence ``seq_of_parameters``. @@ -196,30 +234,15 @@ def executemany(self, operation, seq_of_parameters): Return values are not defined. """ - values_list = [] - RE_INSERT_VALUES = re.compile( - r"\s*((?:INSERT|REPLACE)\s.+\sVALUES?\s*)" - + r"(\(\s*(?:%s|%\(.+\)s)\s*(?:,\s*(?:%s|%\(.+\)s)\s*)*\))" - + r"(\s*(?:ON DUPLICATE.*)?);?\s*\Z", - re.IGNORECASE | re.DOTALL, - ) - - m = RE_INSERT_VALUES.match(operation) - if m: - try: - q_prefix = m.group(1) - q_values = m.group(2).rstrip() - - for parameters in seq_of_parameters: - values_list.append(q_values % _escaper.escape_args(parameters)) - query = "{} {};".format(q_prefix, ",".join(values_list)) + try: + query = self.render_multi_row_insert(operation, seq_of_parameters) + if query is not None: return self.inner.execute(query) - except Exception as e: - # We have to raise dbAPI error - raise Error(str(e)) from e - else: - for parameters in seq_of_parameters: - self.execute(operation, parameters) + except Exception as e: + # We have to raise dbAPI error + raise Error(str(e)) from e + for parameters in seq_of_parameters: + self.execute(operation, parameters) def fetchone(self): """Fetch the next row of a query result set, returning a single sequence, or ``None`` when diff --git a/databend_sqlalchemy/databend_dialect.py b/databend_sqlalchemy/databend_dialect.py index 9af3ef3..6afd1e3 100644 --- a/databend_sqlalchemy/databend_dialect.py +++ b/databend_sqlalchemy/databend_dialect.py @@ -71,6 +71,8 @@ from .types import DOUBLE from sqlalchemy.engine import ExecutionContext, default +from sqlalchemy.engine.interfaces import Dialect +from sqlalchemy.engine.url import URL from sqlalchemy.exc import DBAPIError, NoSuchTableError from .dml import ( @@ -1462,6 +1464,14 @@ def import_dbapi(cls): import connector return connector + @classmethod + def get_async_dialect_cls(cls, url: URL) -> type[Dialect]: + """Lets ``create_async_engine`` take a plain ``databend://`` URL. The import is + deferred because the async dialect module builds on this one.""" + from databend_sqlalchemy.async_dialect import AsyncDatabendDialect + + return AsyncDatabendDialect + def _get_server_version_info(self, connection): val = connection.scalar(text("SELECT VERSION()")) m = re.match(r"(?:.*)v(\d+).(\d+).(\d+)-([^\(]+)(?:\()", val) diff --git a/databend_sqlalchemy/requirements.py b/databend_sqlalchemy/requirements.py index acee519..d7cd19b 100644 --- a/databend_sqlalchemy/requirements.py +++ b/databend_sqlalchemy/requirements.py @@ -55,6 +55,19 @@ def temporary_tables(self): def temp_table_reflection(self): return exclusions.closed() + @property + def async_dialect(self): + """dialect makes use of await_() within and needs to be called inside of a + greenlet""" + return exclusions.only_if(lambda config: config.db.dialect.is_async) + + @property + def buffered_cursors(self): + """target driver can buffer a result in full rather than stream it, which the + asyncio dialect does unless ``stream_results`` is requested; the blocking driver + cursor always streams.""" + return exclusions.only_if(lambda config: config.db.dialect.is_async) + @property def self_referential_foreign_keys(self): """Target database must support self-referential foreign keys.""" diff --git a/pyproject.toml b/pyproject.toml index fcda24e..4b66671 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,74 @@ [build-system] -requires = ["setuptools>=42", "wheel"] +requires = ["setuptools>=77"] build-backend = "setuptools.build_meta" +[project] +name = "databend_sqlalchemy" +authors = [ + {name = "Databend Cloud", email = "hanshanjie@databend.com"}, +] +maintainers = [ + {name = "Databend Cloud", email = "hanshanjie@databend.com"}, +] +description = "Sqlalchemy adapter for Databend" +requires-python = ">=3.10" +keywords = ["databend", "sqlalchemy", "dialect"] +license = "Apache-2.0" +license-files = ["LICENSE"] +classifiers = [ + "Operating System :: OS Independent", + "Intended Audience :: Developers", + "Programming Language :: Python", + "Programming Language :: Python :: 3 :: Only", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", + "Programming Language :: Python :: 3.15", + "Topic :: Database :: Front-Ends", +] +dependencies = [ + "databend-driver>=0.26.2", + "sqlalchemy>=1.4", +] +dynamic = ["readme", "version"] + +[project.optional-dependencies] +asyncio = [ + "sqlalchemy[asyncio]>=2.0.28", +] +dev = [ + "devtools==0.7.0", + "mock==4.0.3", + "pre-commit==2.15.0", + "pytest==8.1.1", + "pytest-cov==3.0.0", + "pytest-xdist==3.5.0", + "sqlalchemy-stubs==0.4", +] +superset = [ + "apache-superset>=1.4.1", +] + +[project.urls] +Homepage = "https://databend.com" +Documentation = "https://docs.databend.com" +Repository = "https://github.com/databendlabs/databend-sqlalchemy" +Issues = "https://github.com/databendlabs/databend-sqlalchemy/issues" + +[project.entry-points."sqlalchemy.dialects"] +databend = "databend_sqlalchemy.databend_dialect:DatabendDialect" +"databend.databend" = "databend_sqlalchemy.databend_dialect:DatabendDialect" +"databend.async" = "databend_sqlalchemy.async_dialect:AsyncDatabendDialect" + +[tool.setuptools.dynamic] +readme = {file = ["README.rst"], content-type = "text/x-rst"} +version = {attr = "databend_sqlalchemy.VERSION"} + +[tool.setuptools.packages.find] +include = ["databend_sqlalchemy*"] + # configure isort to be compatible with black # source: https://black.readthedocs.io/en/stable/compatible_configs.html#configuration [tool.isort] @@ -19,3 +86,7 @@ exclude = ''' | dist )/ ''' + +[tool.pytest.ini_options] +addopts = "--tb native -n4 --dist loadgroup -v -r fxX --maxfail=25 -p no:warnings" +python_files = "tests/*test_*.py" diff --git a/setup.cfg b/setup.cfg index 613cd7e..47f4854 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,58 +1,3 @@ -[metadata] -name = databend_sqlalchemy -version = attr: databend_sqlalchemy.VERSION -description = Sqlalchemy adapter for Databend -long_description = file: README.rst -long_description_content_type = text/markdown -url = https://github.com/databendlabs/databend-sqlalchemy -author = Databend Cloud -author_email = hanshanjie@databend.com -license = Apache-2.0 -license_file = LICENSE -classifiers = - License :: OSI Approved :: Apache Software License - License :: OSI Approved :: MIT License - Operating System :: OS Independent - Programming Language :: Python :: 3 - Programming Language :: Python :: 3 :: Only - Programming Language :: Python :: 3.7 - Programming Language :: Python :: 3.8 - Programming Language :: Python :: 3.9 - Programming Language :: Python :: 3.10 - Programming Language :: Python :: 3.11 - Programming Language :: Python :: 3.12 -project_urls = - Bug Tracker = https://github.com/databendlabs/databend-sqlalchemy - -[options] -packages = find: -install_requires = - databend-driver>=0.26.2 - sqlalchemy>=1.4 -python_requires = >=3.8 -package_dir = - = . - -[options.packages.find] -where = . - -[options.entry_points] -sqlalchemy.dialects = - databend = databend_sqlalchemy.databend_dialect:DatabendDialect - databend.databend = databend_sqlalchemy.databend_dialect:DatabendDialect - -[options.extras_require] -dev = - devtools==0.7.0 - mock==4.0.3 - pre-commit==2.15.0 - pytest==8.1.1 - pytest-cov==3.0.0 - pytest-xdist==3.5.0 - sqlalchemy-stubs==0.4 - -superset = apache_superset>=1.4.1 - [flake8] max-line-length = 88 per-file-ignores = __init__.py:F401 @@ -60,15 +5,10 @@ ignore = E203, W503 ban-relative-imports = True inline-quotes = " -[tool:pytest] -addopts= --tb native -n4 -v -r fxX --maxfail=25 -p no:warnings -python_files=tests/*test_*.py -# python_functions=test_round_trip_same_named_column -# log_level=DEBUG - [sqla_testing] requirement_cls=databend_sqlalchemy.requirements:Requirements profile_file = .profiles.txt [db] default=databend://databend:databend@localhost:8000/default?sslmode=disable +async=databend+async://databend:databend@localhost:8000/default?sslmode=disable diff --git a/tests/conftest.py b/tests/conftest.py index 6676d18..22053da 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,26 +1,51 @@ +from packaging import version from sqlalchemy.dialects import registry +from sqlalchemy.testing.plugin import pytestplugin import pytest +import sqlalchemy registry.register("databend.databend", "databend_sqlalchemy.databend_dialect", "DatabendDialect") registry.register("databend", "databend_sqlalchemy.databend_dialect", "DatabendDialect") +registry.register("databend.async", "databend_sqlalchemy.async_dialect", "AsyncDatabendDialect") pytest.register_assert_rewrite("sa.testing.assertions") from sqlalchemy.testing.plugin.pytestplugin import * -from packaging import version -import sqlalchemy -if version.parse(sqlalchemy.__version__) >= version.parse('2.0.0'): - from sqlalchemy import event, text - from sqlalchemy import Engine - - @event.listens_for(Engine, "connect") - def receive_engine_connect(conn, r): - cur = conn.cursor() - cur.execute('SET global format_null_as_str = 0') - cur.execute('SET global enable_geo_create_table = 1') - try: - cur.execute("SET global sequence_step_size = 1") - except: - pass # don't fail when this setting doesn't exist - cur.close() + +@pytest.hookimpl(tryfirst=True) +def pytest_collection_modifyitems(session, config, items): + """Expand SQLAlchemy's per-backend test classes before pytest-xdist reads the + ``xdist_group`` markers for ``--dist loadgroup``; xdist would otherwise tag the + items SQLAlchemy is about to replace.""" + pytestplugin.pytest_collection_modifyitems(session, config, items) + + +def modules_beyond_installed_sqlalchemy(): + """The compliance suite relies on test requirements SQLAlchemy added in 2.0.42 and the + asyncio dialect needs 2.0.28, so neither module is collected on older releases, which + still run the remaining tests, SQLAlchemy 1.4 included.""" + installed = version.parse(sqlalchemy.__version__) + minimums = {"test_sqlalchemy.py": "2.0.42", "test_async.py": "2.0.28"} + return [ + module + for module, minimum in minimums.items() + if installed < version.parse(minimum) + ] + + +collect_ignore = modules_beyond_installed_sqlalchemy() + +from sqlalchemy import event, text +from sqlalchemy.engine import Engine + +@event.listens_for(Engine, "connect") +def receive_engine_connect(conn, r): + cur = conn.cursor() + cur.execute('SET global format_null_as_str = 0') + cur.execute('SET global enable_geo_create_table = 1') + try: + cur.execute("SET global sequence_step_size = 1") + except: + pass # don't fail when this setting doesn't exist + cur.close() diff --git a/tests/test_async.py b/tests/test_async.py new file mode 100644 index 0000000..c9a84f6 --- /dev/null +++ b/tests/test_async.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python + +from sqlalchemy import Column, Integer, String, Table, func, insert, select, text +from sqlalchemy.exc import DBAPIError +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column +from sqlalchemy.testing import ( + async_test, + config, + eq_, + expect_raises, + fixtures, + is_not_none, + is_true, +) + +from databend_sqlalchemy import async_connector +from databend_sqlalchemy.async_dialect import AsyncDatabendDialect + + +class AsyncDialectSelectionTest(fixtures.TestBase): + __requires__ = ("asyncio",) + + def test_create_async_engine_accepts_plain_url(self): + engine = create_async_engine(config.db.url.set(drivername="databend")) + is_true(isinstance(engine.dialect, AsyncDatabendDialect)) + + +class AsyncEngineTest(fixtures.TablesTest): + """Exercises the asyncio API itself, so it runs under ``--db async``, where the + harness disposes of async engines inside a greenlet.""" + + __backend__ = True + __requires__ = ("async_dialect",) + + @classmethod + def define_tables(cls, metadata): + Table( + "async_users", + metadata, + Column("id", Integer), + Column("name", String(50)), + ) + + @async_test + async def test_buffered_result_outlives_connection(self, async_testing_engine): + users = self.tables.async_users + engine = async_testing_engine() + async with engine.begin() as conn: + await conn.execute( + insert(users), [{"id": 1, "name": "a"}, {"id": 2, "name": "b"}] + ) + result = await conn.execute(select(users).order_by(users.c.id)) + eq_(result.all(), [(1, "a"), (2, "b")]) + + @async_test + async def test_insert_rowcount(self, async_testing_engine): + users = self.tables.async_users + engine = async_testing_engine() + async with engine.begin() as conn: + single = await conn.execute(insert(users).values(id=1, name="a")) + many = await conn.execute( + insert(users), [{"id": i, "name": f"n{i}"} for i in range(2, 5)] + ) + eq_(single.rowcount, 1) + eq_(many.rowcount, 3) + + @async_test + async def test_stream_fetches_rows_as_consumed(self, async_testing_engine): + engine = async_testing_engine() + async with engine.connect() as conn: + result = await conn.stream(text("SELECT number FROM numbers(2500)")) + driver_cursor = result._real_result.cursor._cursor + partitions = result.partitions(1000) + partition_sizes = [len(await anext(partitions))] + is_not_none(driver_cursor._rows) + partition_sizes += [len(p) async for p in partitions] + eq_(partition_sizes, [1000, 1000, 500]) + + @async_test + async def test_session_state_shared_across_statements(self, async_testing_engine): + engine = async_testing_engine() + async with engine.connect() as conn: + await conn.execute( + text("CREATE OR REPLACE TEMPORARY TABLE async_tmp (a INT)") + ) + await conn.execute(text("INSERT INTO async_tmp VALUES (1), (2)")) + eq_(await conn.scalar(text("SELECT count(*) FROM async_tmp")), 2) + + @async_test + async def test_driver_errors_raise_dbapi_error(self, async_testing_engine): + engine = async_testing_engine() + async with engine.connect() as conn: + with expect_raises(DBAPIError): + await conn.execute(text("SELECT * FROM async_no_such_table")) + eq_(await conn.scalar(text("SELECT 1")), 1) + + @async_test + async def test_async_creator(self, async_testing_engine): + dialect = async_testing_engine().dialect + _, connect_args = dialect.create_connect_args(config.db.url) + engine = create_async_engine( + "databend+async://", + async_creator=lambda: async_connector.connect(**connect_args), + ) + try: + async with engine.connect() as conn: + eq_(await conn.scalar(text("SELECT 42")), 42) + finally: + await engine.dispose() + + @async_test + async def test_orm_session_round_trip(self, async_testing_engine): + class Base(DeclarativeBase): + pass + + class Note(Base): + __tablename__ = "async_notes" + id: Mapped[int] = mapped_column(primary_key=True) + body: Mapped[str] = mapped_column(String(50)) + + engine = async_testing_engine() + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + try: + async with AsyncSession(engine) as session: + session.add_all([Note(id=1, body="first"), Note(id=2, body="second")]) + await session.commit() + async with AsyncSession(engine) as session: + notes = (await session.scalars(select(Note).order_by(Note.id))).all() + eq_([(n.id, n.body) for n in notes], [(1, "first"), (2, "second")]) + eq_(await session.scalar(select(func.count()).select_from(Note)), 2) + finally: + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.drop_all) diff --git a/tests/test_basic_usage.py b/tests/test_basic_usage.py new file mode 100644 index 0000000..5328dbb --- /dev/null +++ b/tests/test_basic_usage.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python + +import datetime +import decimal + +from sqlalchemy import ( + Boolean, + Column, + Date, + Float, + Integer, + Numeric, + String, + Table, + inspect, + insert, + select, + text, +) +from sqlalchemy.testing import eq_, fixtures, is_false, is_true + + +class BasicUsageTest(fixtures.TablesTest): + """Covers what SQL tools such as Superset rely on. Unlike the compliance suite, this + runs on every supported SQLAlchemy release, SQLAlchemy 1.4 included.""" + + __backend__ = True + + @classmethod + def define_tables(cls, metadata): + Table( + "basic_usage", + metadata, + Column("id", Integer), + Column("name", String(50)), + Column("score", Float), + Column("active", Boolean), + Column("joined", Date), + Column("balance", Numeric(10, 2)), + ) + + def test_inspector_describes_table(self, connection): + inspector = inspect(connection) + is_true("basic_usage" in inspector.get_table_names()) + is_true(inspector.has_table("basic_usage")) + is_false(inspector.has_table("basic_usage_missing")) + eq_( + [column["name"] for column in inspector.get_columns("basic_usage")], + ["id", "name", "score", "active", "joined", "balance"], + ) + + def test_inspector_describes_view(self, connection): + connection.exec_driver_sql( + "CREATE OR REPLACE VIEW basic_usage_view AS SELECT id, name FROM basic_usage" + ) + try: + inspector = inspect(connection) + is_true("basic_usage_view" in inspector.get_view_names()) + eq_( + [column["name"] for column in inspector.get_columns("basic_usage_view")], + ["id", "name"], + ) + finally: + connection.exec_driver_sql("DROP VIEW IF EXISTS basic_usage_view") + + def test_executemany_then_limit_offset(self, connection): + table = self.tables.basic_usage + connection.execute( + insert(table), [{"id": i, "name": f"n{i}"} for i in range(1, 6)] + ) + rows = connection.execute( + select(table.c.id).order_by(table.c.id).limit(2).offset(1) + ).all() + eq_(rows, [(2,), (3,)]) + + def test_text_with_percent_and_bound_parameter(self, connection): + eq_(connection.scalar(text("SELECT 'a%b'")), "a%b") + eq_( + connection.scalar(text("SELECT CONCAT(:value, '%')"), {"value": "50"}), + "50%", + ) + + def test_types_round_trip(self, connection): + table = self.tables.basic_usage + row = { + "id": 1, + "name": "it's", + "score": 1.5, + "active": True, + "joined": datetime.date(2026, 10, 23), + "balance": decimal.Decimal("12.34"), + } + connection.execute(insert(table).values(**row)) + eq_(connection.execute(select(table)).one()._asdict(), row) diff --git a/tests/test_copy_into.py b/tests/test_copy_into.py index c679510..ea74995 100644 --- a/tests/test_copy_into.py +++ b/tests/test_copy_into.py @@ -30,6 +30,8 @@ import sqlalchemy from packaging import version +ROW_COUNTS_REPORTED = version.parse(sqlalchemy.__version__) >= version.parse('2.0.28') + class CompileDatabendCopyIntoTableTest(fixtures.TestBase, AssertsCompiledSQL): @@ -217,52 +219,53 @@ def define_tables(cls, metadata): Column("data", String(50)), ) - if version.parse(sqlalchemy.__version__) >= version.parse('2.0.0'): - def test_copy_into_stage_and_table(self, connection): - # create stage - connection.execute(text('CREATE OR REPLACE STAGE mystage')) - # copy into stage from random table limiting 1000 - table = self.tables.random_data - query = table.select().limit(1000) + def test_copy_into_stage_and_table(self, connection): + # create stage + connection.execute(text('CREATE OR REPLACE STAGE mystage')) + # copy into stage from random table limiting 1000 + table = self.tables.random_data + query = table.select().limit(1000) - copy_into = CopyIntoLocation( - target=StageClause( - name='mystage' - ), - from_=query, - file_format=ParquetFormat(), - options=CopyIntoLocationOptions() - ) - r = connection.execute( - copy_into - ) + copy_into = CopyIntoLocation( + target=StageClause( + name='mystage' + ), + from_=query, + file_format=ParquetFormat(), + options=CopyIntoLocationOptions() + ) + r = connection.execute( + copy_into + ) + if ROW_COUNTS_REPORTED: eq_(r.rowcount, 1000) - copy_into_results = r.context.copy_into_location_results() - eq_(copy_into_results['rows_unloaded'], 1000) - # eq_(copy_into_results['input_bytes'], 16250) # input bytes will differ, the table is random - # eq_(copy_into_results['output_bytes'], 4701) # output bytes differs + copy_into_results = r.context.copy_into_location_results() + eq_(copy_into_results['rows_unloaded'], 1000) + # eq_(copy_into_results['input_bytes'], 16250) # input bytes will differ, the table is random + # eq_(copy_into_results['output_bytes'], 4701) # output bytes differs - # now copy into table + # now copy into table - copy_into_table = CopyIntoTable( - target=self.tables.loaded, - from_=StageClause( - name='mystage' - ), - file_format=ParquetFormat(), - options=CopyIntoTableOptions() - ) - r = connection.execute( - copy_into_table - ) + copy_into_table = CopyIntoTable( + target=self.tables.loaded, + from_=StageClause( + name='mystage' + ), + file_format=ParquetFormat(), + options=CopyIntoTableOptions() + ) + r = connection.execute( + copy_into_table + ) + if ROW_COUNTS_REPORTED: eq_(r.rowcount, 1000) - copy_into_table_results = r.context.copy_into_table_results() - assert len(copy_into_table_results) == 1 - result = copy_into_table_results[0] - assert result['file'].endswith('.parquet') - eq_(result['rows_loaded'], 1000) - eq_(result['errors_seen'], 0) - eq_(result['first_error'], None) - eq_(result['first_error_line'], None) + copy_into_table_results = r.context.copy_into_table_results() + assert len(copy_into_table_results) == 1 + result = copy_into_table_results[0] + assert result['file'].endswith('.parquet') + eq_(result['rows_loaded'], 1000) + eq_(result['errors_seen'], 0) + eq_(result['first_error'], None) + eq_(result['first_error_line'], None) diff --git a/tests/test_sqlalchemy.py b/tests/test_sqlalchemy.py index 080965a..e4519c1 100644 --- a/tests/test_sqlalchemy.py +++ b/tests/test_sqlalchemy.py @@ -3,7 +3,6 @@ from sqlalchemy.testing.suite import * from sqlalchemy.testing.suite import ComponentReflectionTestExtra as _ComponentReflectionTestExtra -from sqlalchemy.testing.suite import DeprecatedCompoundSelectTest as _DeprecatedCompoundSelectTest from sqlalchemy.testing.suite import BooleanTest as _BooleanTest from sqlalchemy.testing.suite import BinaryTest as _BinaryTest from sqlalchemy.testing.suite import CompoundSelectTest as _CompoundSelectTest @@ -14,6 +13,8 @@ from sqlalchemy.testing.suite import QuotedNameArgumentTest as _QuotedNameArgumentTest from sqlalchemy.testing.suite import JoinTest as _JoinTest from sqlalchemy.testing.suite import HasSequenceTest as _HasSequenceTest +from sqlalchemy.testing.suite import HasSequenceTestEmpty as _HasSequenceTestEmpty +from sqlalchemy.testing.suite import SequenceTest as _SequenceTest from sqlalchemy.testing.suite import ServerSideCursorsTest as _ServerSideCursorsTest @@ -21,27 +22,19 @@ from sqlalchemy.testing.suite import JSONTest as _JSONTest from sqlalchemy.testing.suite import IntegerTest as _IntegerTest +import pytest from sqlalchemy import types as sql_types from sqlalchemy.testing import config, skip_test -from sqlalchemy import testing, Table, Column, Integer +from sqlalchemy import testing, Table, Column, Integer, select, text from sqlalchemy.testing import eq_, fixtures, assertions from databend_sqlalchemy.types import TINYINT, BITMAP, DOUBLE, GEOMETRY, GEOGRAPHY from packaging import version import sqlalchemy -if version.parse(sqlalchemy.__version__) >= version.parse('2.0.0'): - if version.parse(sqlalchemy.__version__) < version.parse('2.0.42'): - from sqlalchemy.testing.suite import BizarroCharacterFKResolutionTest as _BizarroCharacterFKResolutionTest - from sqlalchemy.testing.suite import EnumTest as _EnumTest -else: - from sqlalchemy.testing.suite import ComponentReflectionTest as _ComponentReflectionTest - - class ComponentReflectionTest(_ComponentReflectionTest): - - @testing.skip("databend") - def test_get_indexes(self): - pass +if version.parse(sqlalchemy.__version__) < version.parse('2.0.42'): + from sqlalchemy.testing.suite import BizarroCharacterFKResolutionTest as _BizarroCharacterFKResolutionTest +from sqlalchemy.testing.suite import EnumTest as _EnumTest class ComponentReflectionTestExtra(_ComponentReflectionTestExtra): @testing.skip("databend") #ToDo No length in Databend @@ -127,18 +120,21 @@ def test_limit_offset_in_unions_from_alias(self): pass -class DeprecatedCompoundSelectTest(_DeprecatedCompoundSelectTest): - @testing.skip("databend") - def test_limit_offset_aliased_selectable_in_unions(self): - pass +if version.parse(sqlalchemy.__version__) < version.parse('2.1.0'): + from sqlalchemy.testing.suite import DeprecatedCompoundSelectTest as _DeprecatedCompoundSelectTest - @testing.skip("databend") - def test_limit_offset_selectable_in_unions(self): - pass + class DeprecatedCompoundSelectTest(_DeprecatedCompoundSelectTest): + @testing.skip("databend") + def test_limit_offset_aliased_selectable_in_unions(self): + pass - @testing.skip("databend") - def test_limit_offset_in_unions_from_alias(self): - pass + @testing.skip("databend") + def test_limit_offset_selectable_in_unions(self): + pass + + @testing.skip("databend") + def test_limit_offset_in_unions_from_alias(self): + pass class HasIndexTest(_HasIndexTest): @@ -228,7 +224,7 @@ def test_get_indexes(self, name): class JoinTest(_JoinTest): __requires__ = ("foreign_keys",) -if version.parse(sqlalchemy.__version__) >= version.parse('2.0.0') and version.parse(sqlalchemy.__version__) < version.parse('2.0.42'): +if version.parse(sqlalchemy.__version__) < version.parse('2.0.42'): class BizarroCharacterFKResolutionTest(_BizarroCharacterFKResolutionTest): __requires__ = ("foreign_keys",) @@ -253,29 +249,36 @@ def test_pickle_roundtrip(self): class ServerSideCursorsTest(_ServerSideCursorsTest): def _is_server_side(self, cursor): - # ToDo - requires implementation of `stream_results` option, so True always for now - if self.engine.dialect.driver == "databend": - return True - return super() + """The blocking driver cursor always streams, so only the asyncio dialect, which buffers + unless ``stream_results`` is requested, has cursors that are not server side.""" + if self.engine.dialect.is_async: + return cursor.server_side + return True - # ToDo - The commented out testing combinations here should be reviewed when `stream_results` is implemented @testing.combinations( ("global_string", True, "select 1", True), ("global_text", True, text("select 1"), True), ("global_expr", True, select(1), True), - # ("global_off_explicit", False, text("select 1"), False), + ( + "global_off_explicit", + False, + text("select 1"), + False, + testing.requires.buffered_cursors, + ), ( "stmt_option", False, select(1).execution_options(stream_results=True), True, ), - # ( - # "stmt_option_disabled", - # True, - # select(1).execution_options(stream_results=False), - # False, - # ), + ( + "stmt_option_disabled", + True, + select(1).execution_options(stream_results=False), + False, + testing.requires.buffered_cursors, + ), ("for_update_expr", True, select(1).with_for_update(), True), # TODO: need a real requirement for this, or dont use this test # ( @@ -285,7 +288,13 @@ def _is_server_side(self, cursor): # True, # testing.skip_if(["sqlite", "mssql"]), # ), - # ("text_no_ss", False, text("select 42"), False), + ( + "text_no_ss", + False, + text("select 42"), + False, + testing.requires.buffered_cursors, + ), ( "text_ss_option", False, @@ -298,15 +307,24 @@ def _is_server_side(self, cursor): def test_ss_cursor_status( self, engine_ss_arg, statement, cursor_ss_status ): - super() - - @testing.skip("databend") # ToDo - requires implementation of `stream_results` option + """Repeats the base test, which can't be reached through ``super()`` as its own + combinations wrap it to take an extra exclusions argument.""" + engine = self._fixture(engine_ss_arg) + with engine.begin() as conn: + if isinstance(statement, str): + result = conn.exec_driver_sql(statement) + else: + result = conn.execute(statement) + eq_(self._is_server_side(result.cursor), cursor_ss_status) + result.close() + + @testing.requires.buffered_cursors def test_stmt_enabled_conn_option_disabled(self): - pass + super().test_stmt_enabled_conn_option_disabled() - @testing.skip("databend") # ToDo - requires implementation of `stream_results` option + @testing.requires.buffered_cursors def test_aliases_and_ss(self): - pass + super().test_aliases_and_ss() @testing.skip("databend") # Skipped because requires auto increment primary key def test_roundtrip_fetchall(self): @@ -316,13 +334,12 @@ def test_roundtrip_fetchall(self): def test_roundtrip_fetchmany(self): pass -if version.parse(sqlalchemy.__version__) >= version.parse('2.0.0'): - class EnumTest(_EnumTest): - __backend__ = True +class EnumTest(_EnumTest): + __backend__ = True - @testing.skip("databend") # Skipped because no supporting enums yet - def test_round_trip_executemany(self, connection): - pass + @testing.skip("databend") # Skipped because no supporting enums yet + def test_round_trip_executemany(self, connection): + pass class CTETest(_CTETest): @@ -688,6 +705,22 @@ def test_geography_write_and_read(self, connection): eq_(result, ('{"type": "GeometryCollection", "geometries": [{"type": "Point", "coordinates": [10,20]},{"type": "LineString", "coordinates": [[10,20],[30,40]]},{"type": "Polygon", "coordinates": [[[10,20],[30,40],[50,60],[10,20]]]}]}')) +databend_sequences = pytest.mark.xdist_group("databend_sequences") + + +@databend_sequences +class SequenceTest(_SequenceTest): + """Databend sequences belong to the tenant rather than to a database, so each xdist + worker's database would share them. The sequence tests join one xdist group, which + ``--dist loadgroup`` runs on a single worker, rather than racing on the same names.""" + + +@databend_sequences +class HasSequenceTestEmpty(_HasSequenceTestEmpty): + """Expects no sequences to exist, so it runs alongside the other sequence tests.""" + + +@databend_sequences class HasSequenceTest(_HasSequenceTest): # ToDo - overridden other_seq definition due to lack of sequence ddl support for nominvalue nomaxvalue