diff --git a/CHANGELOG.md b/CHANGELOG.md
index 65e9054..95f19e8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,11 +2,12 @@
 
 ## [Unreleased]
 
-- Switch packaging and project workflows to poetry
-- Drop support for python 3.7
-- Drop support for Exasol 6.x
-- Drop support for Exasol 7.0.x
-- Relock dependencies (Internal)
+- Added dbapi2 compliant driver interface `exasol.driver.websocket` ontop of pyexasol
+- Switched packaging and project workflow to poetry
+- Droped support for python 3.7
+- Droped support for Exasol 6.x
+- Droped support for Exasol 7.0.x
+- Relocked dependencies (Internal)
 
 ## [0.25.2] - 2023-03-14
 
diff --git a/README.md b/README.md
index 4acdfe9..0276aad 100644
--- a/README.md
+++ b/README.md
@@ -54,7 +54,7 @@ PyEXASOL provides API to read & write multiple data streams in parallel using se
 ## System requirements
 
 - Exasol >= 7.1
-- Python >= 3.8
+- Python >= 3.9
 
 
 ## Getting started
diff --git a/docs/DBAPI_COMPAT.md b/docs/DBAPI_COMPAT.md
index 38ad479..526a0cf 100644
--- a/docs/DBAPI_COMPAT.md
+++ b/docs/DBAPI_COMPAT.md
@@ -1,8 +1,28 @@
 ## DB-API 2.0 compatibility
 
-PyEXASOL [public interface](/docs/REFERENCE.md) is similar to [PEP-249 DB-API 2.0](https://www.python.org/dev/peps/pep-0249/) specification, but it does not strictly follow it. This page explains the reasons behind this decision.
+PyEXASOL [public interface](/docs/REFERENCE.md) is similar to [PEP-249 DB-API 2.0](https://www.python.org/dev/peps/pep-0249/) specification, but it does not strictly follow it. This page explains the reasons behind this decision and your alternative(s) if you need or want to use a dbabpi2 compatible driver..
 
-If you absolutely need DB-API 2.0 compatibility, you may use [TurbODBC](https://github.com/blue-yonder/turbodbc) instead.
+### Alternatives
+
+#### Exasol WebSocket Driver
+The `pyexasol` package includes a DBAPI2 compatible driver facade, located in the `exasol.driver` package. However, using pyexasol directly will generally yield better performance when utilizing Exasol in an OLAP manner, which is likely the typical use case. 
+
+That said, in specific scenarios, the DBAPI2 API can be advantageous or even necessary. This is particularly true when integrating with "DB-Agnostic" frameworks. In such cases, you can just import and use the DBAPI2 compliant facade as illustrated in the example below.
+
+```python
+from exasol.driver.websocket.dbapi2 import connect
+
+connection = connect(dsn='', username='sys', password='exasol', schema='TEST')
+
+with connection.cursor() as cursor:
+    cursor.execute("SELECT 1;")
+```
+
+#### TurboODBC
+[TurboODBC](https://github.com/blue-yonder/turbodbc) offers an alternative ODBC-based, DBAPI2-compatible driver, which supports the Exasol database.
+
+#### Pyodbc
+[Pyodbc](https://github.com/mkleehammer/pyodbc) provides an ODBC-based, DBAPI2-compatible driver. For further details, please refer to our [wiki](https://github.com/mkleehammer/pyodbc/wiki).
 
 ### Rationale
 
@@ -66,32 +86,3 @@ Replace with:
 C.export_to_pandas('SELECT * FROM table')
 ```
 etc.
-
-## DB-API 2.0 wrapper
-
-In order to make it easier to start using PyEXASOL, simple DB-API 2.0 wrapper is provided. It works for `SELECT` statements without placeholders. Please see the example:
-
-```python
-# Import "wrapped" version of PyEXASOL module
-import pyexasol.db2 as E
-
-C = E.connect(dsn=config.dsn, user=config.user, password=config.password, schema=config.schema)
-
-# Cursor
-cur = C.cursor()
-cur.execute("SELECT * FROM users ORDER BY user_id LIMIT 5")
-
-# Standard .description and .rowcount attributes
-print(cur.description)
-print(cur.rowcount)
-
-# Standard fetching
-while True:
-    row = cur.fetchone()
-
-    if row is None:
-        break
-
-    print(row)
-
-```
diff --git a/exasol/driver/__init__.py b/exasol/driver/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/exasol/driver/websocket/__init__.py b/exasol/driver/websocket/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/exasol/driver/websocket/_connection.py b/exasol/driver/websocket/_connection.py
new file mode 100644
index 0000000..19e6e16
--- /dev/null
+++ b/exasol/driver/websocket/_connection.py
@@ -0,0 +1,165 @@
+"""
+This module provides `PEP-249`_ DBAPI compliant connection implementation.
+(see also `PEP-249-connection`_)
+
+.. _PEP-249-connection: https://peps.python.org/pep-0249/#connection-objects
+"""
+
+import ssl
+from functools import wraps
+
+import pyexasol
+
+from exasol.driver.websocket._cursor import Cursor as DefaultCursor
+from exasol.driver.websocket._errors import Error
+
+
+def _requires_connection(method):
+    """
+    Decorator requires the object to have a working connection.
+
+    Raises:
+        Error if the connection object has no active connection.
+    """
+
+    @wraps(method)
+    def wrapper(self, *args, **kwargs):
+        if not self._connection:
+            raise Error("No active connection available")
+        return method(self, *args, **kwargs)
+
+    return wrapper
+
+
+class Connection:
+    """
+    Implementation of a websocket-based connection.
+
+    For more details see :class: `Connection` protocol definition.
+    """
+
+    def __init__(
+        self,
+        dsn: str = None,
+        username: str = None,
+        password: str = None,
+        schema: str = "",
+        autocommit: bool = True,
+        tls: bool = True,
+        certificate_validation: bool = True,
+        client_name: str = "EXASOL:DBAPI2:WS",
+        client_version: str = "unknown",
+    ):
+        """
+        Create a Connection object.
+
+        Args:
+
+            dsn: Connection string, same format as for standard JDBC / ODBC drivers.
+            username: which will be used for the authentication.
+            password: which will be used for the authentication.
+            schema: to open after connecting.
+            autocommit: enable autocommit.
+            tls: enable tls.
+            certificate_validation: disable certificate validation.
+            client_name: which is communicated to the DB server.
+        """
+
+        # for more details see pyexasol.connection.ExaConnection
+        self._options = {
+            "dsn": dsn,
+            "user": username,
+            "password": password,
+            "schema": schema,
+            "autocommit": autocommit,
+            "snapshot_transactions": None,
+            "connection_timeout": 10,
+            "socket_timeout": 30,
+            "query_timeout": 0,
+            "compression": False,
+            "encryption": tls,
+            "fetch_dict": False,
+            "fetch_mapper": None,
+            "fetch_size_bytes": 5 * 1024 * 1024,
+            "lower_ident": False,
+            "quote_ident": False,
+            "json_lib": "json",
+            "verbose_error": True,
+            "debug": False,
+            "debug_logdir": None,
+            "udf_output_bind_address": None,
+            "udf_output_connect_address": None,
+            "udf_output_dir": None,
+            "http_proxy": None,
+            "client_name": client_name,
+            "client_version": client_version,
+            "protocol_version": 3,
+            "websocket_sslopt": (
+                {"cert_reqs": ssl.CERT_REQUIRED} if certificate_validation else None
+            ),
+            "access_token": None,
+            "refresh_token": None,
+        }
+        self._connection = None
+
+    def connect(self):
+        """See also :py:meth: `Connection.connect`"""
+        try:
+            self._connection = pyexasol.connect(**self._options)
+        except pyexasol.exceptions.ExaConnectionError as ex:
+            raise Error(f"Connection failed, {ex}") from ex
+        except Exception as ex:
+            raise Error() from ex
+        return self
+
+    @property
+    def connection(self):
+        """Underlying connection used by this Connection"""
+        return self._connection
+
+    def close(self):
+        """See also :py:meth: `Connection.close`"""
+        connection_to_close = self._connection
+        self._connection = None
+        if connection_to_close is None or connection_to_close.is_closed:
+            return
+        try:
+            connection_to_close.close()
+        except Exception as ex:
+            raise Error() from ex
+
+    @_requires_connection
+    def commit(self):
+        """See also :py:meth: `Connection.commit`"""
+        try:
+            self._connection.commit()
+        except Exception as ex:
+            raise Error() from ex
+
+    @_requires_connection
+    def rollback(self):
+        """See also :py:meth: `Connection.rollback`"""
+        try:
+            self._connection.rollback()
+        except Exception as ex:
+            raise Error() from ex
+
+    @_requires_connection
+    def cursor(self):
+        """See also :py:meth: `Connection.cursor`"""
+        return DefaultCursor(self)
+
+    def __del__(self):
+        if self._connection is None:
+            return
+
+        # Currently, the only way to handle this gracefully is to invoke the`__del__`
+        # method of the underlying connection rather than calling an explicit `close`.
+        #
+        # For more details, see also:
+        # * https://github.com/exasol/sqlalchemy-exasol/issues/390
+        # * https://github.com/exasol/pyexasol/issues/108
+        #
+        # If the above tickets are resolved, it should be safe to switch back to using
+        # `close` instead of `__del__`.
+        self._connection.__del__()
diff --git a/exasol/driver/websocket/_cursor.py b/exasol/driver/websocket/_cursor.py
new file mode 100644
index 0000000..8ff9bf5
--- /dev/null
+++ b/exasol/driver/websocket/_cursor.py
@@ -0,0 +1,367 @@
+"""
+This module provides `PEP-249`_ DBAPI compliant cursor implementation.
+(see also `PEP-249-cursor`_)
+
+.. _PEP-249-cursor: https://peps.python.org/pep-0249/#cursor-objects
+"""
+
+import datetime
+import decimal
+from collections import defaultdict
+from dataclasses import (
+    astuple,
+    dataclass,
+)
+from functools import wraps
+from typing import Optional
+
+import pyexasol.exceptions
+
+from exasol.driver.websocket._errors import (
+    Error,
+    NotSupportedError,
+)
+from exasol.driver.websocket._types import TypeCode
+
+
+@dataclass
+class MetaData:
+    """Meta data describing a result column"""
+
+    name: str
+    type_code: TypeCode
+    display_size: Optional[int] = None
+    internal_size: Optional[int] = None
+    precision: Optional[int] = None
+    scale: Optional[int] = None
+    null_ok: Optional[bool] = None
+
+
+def _pyexasol2dbapi_metadata(name, metadata) -> MetaData:
+    type_mapping = {t.value: t for t in TypeCode}
+    key_mapping = {
+        "name": "name",
+        "type_code": "type",
+        "precision": "precision",
+        "scale": "scale",
+        "display_size": "unknown",
+        "internal_size": "size",
+        "null_ok": "unknown",
+    }
+    metadata = defaultdict(lambda: None, metadata)
+    metadata["type"] = type_mapping[metadata["type"]]
+    metadata["name"] = name
+    return MetaData(**{k: metadata[v] for k, v in key_mapping.items()})
+
+
+def _is_not_closed(method):
+    """
+    Mark a function to require an open connection.
+
+    Raises:
+        An Error if the marked function is called without an open connection.
+    """
+
+    @wraps(method)
+    def wrapper(self, *args, **kwargs):
+        if self._is_closed:
+            raise Error(
+                f"Unable to execute operation <{method.__name__}>, because cursor was already closed."
+            )
+        return method(self, *args, **kwargs)
+
+    return wrapper
+
+
+def _requires_result(method):
+    """
+    Decorator requires the object to have a result.
+
+    Raises:
+        Error if the cursor object has not produced a result yet.
+    """
+
+    @wraps(method)
+    def wrapper(self, *args, **kwargs):
+        if not self._cursor:
+            raise Error("No result has been produced.")
+        return method(self, *args, **kwargs)
+
+    return wrapper
+
+
+def _identity(value):
+    return value
+
+
+def _pyexasol2dbapi(value, metadata):
+    members = (
+        "name",
+        "type_code",
+        "display_size",
+        "internal_size",
+        "precision",
+        "scale",
+        "null_ok",
+    )
+    metadata = MetaData(**{k: v for k, v in zip(members, metadata)})
+
+    def to_date(v):
+        if not isinstance(v, str):
+            return v
+        return datetime.date.fromisoformat(v)
+
+    def to_float(v):
+        if not isinstance(v, str):
+            return v
+        return float(v)
+
+    converters = defaultdict(
+        lambda: _identity,
+        {
+            TypeCode.Date: to_date,
+            TypeCode.Double: to_float,
+        },
+    )
+    converter = converters[metadata.type_code]
+    return converter(value)
+
+
+def _dbapi2pyexasol(value):
+    converters = defaultdict(
+        lambda: _identity,
+        {decimal.Decimal: str, float: str, datetime.date: str, datetime.datetime: str},
+    )
+    converter = converters[type(value)]
+    return converter(value)
+
+
+class Cursor:
+    """
+    Implementation of a cursor based on the DefaultConnection.
+
+    For more details see :class: `Cursor` protocol definition.
+    """
+
+    # see https://peps.python.org/pep-0249/#arraysize
+    DBAPI_DEFAULT_ARRAY_SIZE = 1
+
+    def __init__(self, connection):
+        self._connection = connection
+        self._cursor = None
+        self._is_closed = False
+
+    # Note: Needed for compatibility with sqlalchemy_exasol base dialect.
+    def __enter__(self):
+        return self
+
+    # Note: Needed for compatibility with sqlalchemy_exasol base dialect.
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.close()
+
+    @property
+    @_is_not_closed
+    def arraysize(self):
+        """See also :py:meth: `Cursor.arraysize`"""
+        return self.DBAPI_DEFAULT_ARRAY_SIZE
+
+    @property
+    @_is_not_closed
+    def description(self):
+        """See also :py:meth: `Cursor.description`"""
+        if not self._cursor:
+            return None
+        columns_metadata = (
+            _pyexasol2dbapi_metadata(name, metadata)
+            for name, metadata in self._cursor.columns().items()
+        )
+        columns_metadata = tuple(astuple(metadata) for metadata in columns_metadata)
+        columns_metadata = columns_metadata if columns_metadata != () else None
+        return columns_metadata
+
+    @property
+    @_is_not_closed
+    def rowcount(self):
+        """
+        See also :py:meth: `Cursor.rowcount`
+
+        Attention: This implementation of the rowcount attribute deviates slightly
+                   from what the dbapi2 requires.
+
+               Difference:
+
+                    If the rowcount of the last operation cannot be determined it will
+                    return 0.
+
+                    Expected by DBAPI2 = -1
+                    Actually returned  = 0
+
+               Rational:
+
+                    With the usage of pyexasol as underlying driver, there is no trivial
+                    way to do this.
+        """
+        if not self._cursor:
+            return -1
+        return self._cursor.rowcount()
+
+    @_is_not_closed
+    def callproc(self, procname, parameters=None):
+        """See also :py:meth: `Cursor.callproc`"""
+        raise NotSupportedError("Optional and therefore not supported")
+
+    @_is_not_closed
+    def close(self):
+        """See also :py:meth: `Cursor.close`"""
+        self._is_closed = True
+        if not self._cursor:
+            return
+        self._cursor.close()
+
+    @_is_not_closed
+    def execute(self, operation, parameters=None):
+        """See also :py:meth: `Cursor.execute`"""
+        if parameters:
+            self.executemany(operation, [parameters])
+            return
+
+        connection = self._connection.connection
+        try:
+            self._cursor = connection.execute(operation)
+        except pyexasol.exceptions.ExaError as ex:
+            raise Error() from ex
+
+    @staticmethod
+    def _adapt_to_requested_db_types(parameters, db_response):
+        """
+        Adapt parameter types to match the types requested by the DB in the
+        `createPreparedStatement <https://github.com/exasol/websocket-api/blob/master/docs/commands/createPreparedStatementV1.md>`_
+        response.
+
+        Args:
+
+            parameters: which will be passed/send to the database.
+            db_response: contains the DB response including the required types.
+
+
+        Attention:
+
+            This shim method currently only patches the following types:
+
+                * VARCHAR
+                * DOUBLE
+
+            therefore it the future it may be necessary to improve or extend this.
+
+            A hint that patching of a specific type is required, could be and
+            error message similar to this one:
+
+            .. code-block::
+
+                pyexasol.exceptions.ExaRequestError:
+                ...
+                message  => getString: JSON value is not a string. (...)
+                ...
+        """
+
+        def varchar(value):
+            if value is None:
+                return None
+            return str(value)
+
+        def double(value):
+            if value is None:
+                return None
+            return float(value)
+
+        converters = defaultdict(
+            lambda: _identity, {"VARCHAR": varchar, "DOUBLE": double}
+        )
+        selected_converters = (
+            converters[column["dataType"]["type"]] for column in db_response["columns"]
+        )
+        parameters = zip(selected_converters, parameters)
+        parameters = [converter(value) for converter, value in parameters]
+        return parameters
+
+    @_is_not_closed
+    def executemany(self, operation, seq_of_parameters):
+        """See also :py:meth: `Cursor.executemany`"""
+        parameters = [
+            [_dbapi2pyexasol(p) for p in params] for params in seq_of_parameters
+        ]
+        connection = self._connection.connection
+        self._cursor = connection.cls_statement(connection, operation, prepare=True)
+
+        parameter_data = self._cursor.parameter_data
+        parameters = [
+            Cursor._adapt_to_requested_db_types(params, parameter_data)
+            for params in parameters
+        ]
+        try:
+            self._cursor.execute_prepared(parameters)
+        except pyexasol.exceptions.ExaError as ex:
+            raise Error() from ex
+
+    def _convert(self, rows):
+        if rows is None:
+            return None
+
+        return tuple(self._convert_row(row) for row in rows)
+
+    def _convert_row(self, row):
+        if row is None:
+            return row
+
+        return tuple(
+            _pyexasol2dbapi(value, metadata)
+            for value, metadata in zip(row, self.description)
+        )
+
+    @_requires_result
+    @_is_not_closed
+    def fetchone(self):
+        """See also :py:meth: `Cursor.fetchone`"""
+        row = self._cursor.fetchone()
+        return self._convert_row(row)
+
+    @_requires_result
+    @_is_not_closed
+    def fetchmany(self, size=None):
+        """See also :py:meth: `Cursor.fetchmany`"""
+        size = size if size is not None else self.arraysize
+        rows = self._cursor.fetchmany(size)
+        return self._convert(rows)
+
+    @_requires_result
+    @_is_not_closed
+    def fetchall(self):
+        """See also :py:meth: `Cursor.fetchall`"""
+        rows = self._cursor.fetchall()
+        return self._convert(rows)
+
+    @_is_not_closed
+    def nextset(self):
+        """See also :py:meth: `Cursor.nextset`"""
+        raise NotSupportedError("Optional and therefore not supported")
+
+    @_is_not_closed
+    def setinputsizes(self, sizes):
+        """See also :py:meth: `Cursor.setinputsizes`
+
+        Attention:
+            This method does nothing.
+        """
+
+    @_is_not_closed
+    def setoutputsize(self, size, column):
+        """See also :py:meth: `Cursor.setoutputsize`
+
+        Attention:
+            This method does nothing.
+        """
+
+    def __del__(self):
+        if self._is_closed:
+            return
+        self.close()
diff --git a/exasol/driver/websocket/_errors.py b/exasol/driver/websocket/_errors.py
new file mode 100644
index 0000000..b82c08f
--- /dev/null
+++ b/exasol/driver/websocket/_errors.py
@@ -0,0 +1,76 @@
+"""
+This module provides `PEP-249`_ compliant DBAPI exceptions.
+(see also `PEP-249-exceptions`_)
+
+.. _PEP-249-exceptions: https://peps.python.org/pep-0249/#exceptions
+"""
+
+
+class Warning(Exception):  # Required by spec. pylint: disable=W0622
+    """
+    Exception raised for important warnings like data truncations while inserting, etc.
+    """
+
+
+class Error(Exception):
+    """
+    Base class of all other error exceptions.
+    You can use this to catch all errors with one single except statement.
+    Warnings are not considered errors and thus should not use this class as base.
+    """
+
+
+class InterfaceError(Error):
+    """
+    Exception raised for errors that are related to the database interface rather than
+    the database itself.
+    """
+
+
+class DatabaseError(Error):
+    """Exception raised for errors that are related to the database."""
+
+
+class DataError(DatabaseError):
+    """
+    Exception raised for errors that are due to problems with the processed
+    data like division by zero, numeric value out of range, etc.
+    """
+
+
+class OperationalError(DatabaseError):
+    """
+    Exception raised for errors that are related to the database’s operation
+    and not necessarily under the control of the programmer, e.g. an unexpected
+    disconnect occurs, the data source name is not found, a transaction
+    could not be processed, a memory allocation error occurred during processing, etc.
+    """
+
+
+class IntegrityError(DatabaseError):
+    """
+    Exception raised when the relational integrity of the database is affected,
+    e.g. a foreign key check fails.
+    """
+
+
+class InternalError(DatabaseError):
+    """
+    Exception raised when the database encounters an internal error,
+    e.g. the cursor is not valid anymore, the transaction is out of sync, etc.
+    """
+
+
+class ProgrammingError(DatabaseError):
+    """
+    Exception raised for programming errors, e.g. table not found or already exists,
+    syntax error in the SQL statement, wrong number of parameters specified, etc.
+    """
+
+
+class NotSupportedError(DatabaseError):
+    """
+    Exception raised in case a method or database API was used which is not supported
+    by the database, e.g. requesting a .rollback() on a connection that does not
+    support transaction or has transactions turned off.
+    """
diff --git a/exasol/driver/websocket/_protocols.py b/exasol/driver/websocket/_protocols.py
new file mode 100644
index 0000000..b4302b9
--- /dev/null
+++ b/exasol/driver/websocket/_protocols.py
@@ -0,0 +1,273 @@
+"""
+This module provides `PEP-249`_ a DBAPI compliant Connection and Cursor protocol definition.
+(see also `PEP-249-connection`_ and `PEP-249-cursor`_)
+
+.. _PEP-249-connection: https://peps.python.org/pep-0249/#connection-objects
+.. _PEP-249-cursor: https://peps.python.org/pep-0249/#cursor-objects
+"""
+
+from typing import Protocol
+
+
+class Connection(Protocol):
+    """
+    Defines a connection protocol based on `connection-objects`_.
+
+    .. connection-objects: https://peps.python.org/pep-0249/#connection-objects
+    """
+
+    def connect(self):
+        """
+        Connect to the database.
+
+        Attention:
+            Addition not required by PEP-249.
+        """
+
+    def close(self):
+        """
+        Close the connection now (rather than whenever .__del__() is called).
+
+        The connection will be unusable from this point forward; an Error (or subclass)
+        exception will be raised if any operation is attempted with the connection.
+        The same applies to all cursor objects trying to use the connection.
+        Note that closing a connection without committing the changes first will cause
+        an implicit rollback to be performed.
+        """
+
+    def commit(self):
+        """
+        Commit any pending transaction to the database.
+
+        Note:
+            If the database supports an auto-commit feature, this must be initially off.
+            An interface method may be provided to turn it back on. Database modules
+            that do not support transactions should implement this method with
+            void functionality.
+        """
+
+    def rollback(self):
+        """
+        This method is optional since not all databases provide transaction support.
+
+        In case a database does provide transactions this method causes the database
+        to roll back to the start of any pending transaction. Closing a connection
+        without committing the changes first will cause an implicit rollback
+        to be performed.
+        """
+
+    def cursor(self):
+        """
+        Return a new Cursor Object using the connection.
+
+        If the database does not provide a direct cursor concept, the module will have
+        to emulate cursors using other means to the extent needed
+        by this specification.
+        """
+
+
+class Cursor(Protocol):
+    """
+    Defines a protocol which is compliant with `cursor-objects`_.
+
+    .. cursor-objects: https://peps.python.org/pep-0249/#cursor-objects
+    """
+
+    @property
+    def arraysize(self):
+        """
+        This read/write attribute specifies the number of rows to fetch
+        at a time with .fetchmany().
+
+        It defaults to 1, meaning to fetch a single row at a time.
+        Implementations must observe this value with respect to the .fetchmany() method,
+        but are free to interact with the database a single row at a time.
+        It may also be used in the implementation of .executemany().
+        """
+
+    @property
+    def description(self):
+        """
+        This read-only attribute is a sequence of 7-item sequences.
+
+        Each of these sequences contains information describing one result column:
+
+            * name
+            * type_code
+            * display_size
+            * internal_size
+            * precision
+            * scale
+            * null_ok
+
+        The first two items (name and type_code) are mandatory, the other five
+        are optional and are set to None if no meaningful values can be provided.
+
+        This attribute will be None for operations that do not return rows or if
+        the cursor has not had an operation invoked via the .execute*() method yet.
+        """
+
+    @property
+    def rowcount(self):
+        """
+        This read-only attribute specifies the number of rows that the last .execute*()
+        produced (for DQL statements like SELECT) or affected (for DML statements
+        like UPDATE or INSERT).
+
+        The attribute is -1 in case no .execute*() has been performed on the cursor or
+        the rowcount of the last operation cannot be determined by the interface.
+
+        .. note::
+
+            Future versions of the DB API specification could redefine the latter case
+            to have the object return None instead of -1.
+        """
+
+    def callproc(self, procname, parameters):
+        """
+        Call a stored database procedure with the given name.
+        (This method is optional since not all databases provide stored procedures)
+
+        The sequence of parameters must contain one entry for each argument that the
+        procedure expects. The result of the call is returned as a modified copy of
+        the input sequence. Input parameters are left untouched, output and
+        input/output parameters replaced with possibly new values.
+
+        The procedure may also provide a result set as output. This must then be
+        made available through the standard .fetch*() methods.
+        """
+
+    def close(self):
+        """
+        Close the cursor now (rather than whenever __del__ is called).
+
+        The cursor will be unusable from this point forward; an Error (or subclass)
+        exception will be raised if any operation is attempted with the cursor.
+        """
+
+    def execute(self, operation, parameters=None):
+        """
+        Prepare and execute a database operation (query or command).
+
+        Parameters may be provided as sequence or mapping and will be bound to
+        variables in the operation. Variables are specified in a database-specific
+        notation (see the module’s paramstyle attribute for details).
+
+        A reference to the operation will be retained by the cursor.
+        If the same operation object is passed in again, then the cursor can optimize
+        its behavior. This is most effective for algorithms where the same operation
+        is used, but different parameters are bound to it (many times).
+
+        For maximum efficiency when reusing an operation, it is best to use the
+        .setinputsizes() method to specify the parameter types and sizes ahead of time.
+        It is legal for a parameter to not match the predefined information;
+        the implementation should compensate, possibly with a loss of efficiency.
+
+        The parameters may also be specified as list of tuples to e.g. insert multiple
+        rows in a single operation, but this kind of usage is deprecated: .executemany()
+        should be used instead.
+
+        Return values are not defined.
+        """
+
+    def executemany(self, operation, seq_of_parameters):
+        """
+        Prepare a database operation (query or command) and then execute it against all
+        parameter sequences or mappings found in the sequence seq_of_parameters.
+
+        Modules are free to implement this method using multiple calls to the .execute()
+        method or by using array operations to have the database process the sequence
+        as a whole in one call.
+
+        Use of this method for an operation which produces one or more result sets
+        constitutes undefined behavior, and the implementation is permitted
+        (but not required) to raise an exception when it detects that a result set
+        has been created by an invocation of the operation.
+
+        The same comments as for .execute() also apply accordingly to this method.
+
+        Return values are not defined.
+        """
+
+    def fetchone(self):
+        """
+        Fetch the next row of a query result set, returning a single sequence, or None
+        when no more data is available.
+
+        An Error (or subclass) exception is raised if the previous call to .execute*()
+        did not produce any result set or no call was issued yet.
+        """
+
+    def fetchmany(self, size=None):
+        """
+        Fetch the next set of rows of a query result, returning a sequence of sequences
+        (e.g. a list of tuples).
+
+        An empty sequence is returned when no more rows are available. The number of
+        rows to fetch per call is specified by the parameter. If it is not given,
+        the cursor’s arraysize determines the number of rows to be fetched. The method
+        should try to fetch as many rows as indicated by the size parameter.
+        If this is not possible due to the specified number of rows not being available,
+        fewer rows may be returned.
+
+        An Error (or subclass) exception is raised if the previous call to .execute*()
+        did not produce any result set or no call was issued yet.
+
+        Note there are performance considerations involved with the size parameter.
+        For optimal performance, it is usually best to use the .arraysize attribute.
+        If the size parameter is used, then it is best for it to retain the same value
+        from one .fetchmany() call to the next.
+        """
+
+    def fetchall(self):
+        """
+        Fetch all (remaining) rows of a query result, returning them as a sequence of
+        sequences (e.g. a list of tuples).
+
+        Note that the cursor’s arraysize attribute can affect the performance of this
+        operation. An Error (or subclass) exception is raised if the previous call to
+        .execute*() did not produce any result set or no call was issued yet.
+        """
+
+    def nextset(self):
+        """
+        This method will make the cursor skip to the next available set, discarding any
+        remaining rows from the current set.
+        (This method is optional since not all databases support multiple result sets)
+
+        If there are no more sets, the method returns None. Otherwise, it returns a true
+        value and subsequent calls to the .fetch*() methods will return rows from the
+        next result set.
+
+        An Error (or subclass) exception is raised if the previous call to .execute*()
+        did not produce any result set or no call was issued yet.
+        """
+
+    def setinputsizes(self, sizes):
+        """
+        This can be used before a call to .execute*() to predefine memory areas for the
+        operation’s parameters.
+
+        sizes is specified as a sequence — one item for each input parameter. The item
+        should be a Type Object that corresponds to the input that will be used, or it
+        should be an integer specifying the maximum length of a string parameter. If the
+        item is None, then no predefined memory area will be reserved for that column
+        (this is useful to avoid predefined areas for large inputs).
+
+        This method would be used before the .execute*() method is invoked.
+
+        Implementations are free to have this method do nothing and users are free
+        to not use it.
+        """
+
+    def setoutputsizes(self, size, column):
+        """
+        Set a column buffer size for fetches of large columns (e.g. LONGs, BLOBs, etc.).
+
+        The column is specified as an index into the result sequence. Not specifying
+        the column will set the default size for all large columns in the cursor.
+        This method would be used before the .execute*() method is invoked.
+
+        Implementations are free to have this method do nothing and users are free
+        to not use it.
+        """
diff --git a/exasol/driver/websocket/_types.py b/exasol/driver/websocket/_types.py
new file mode 100644
index 0000000..39fd26d
--- /dev/null
+++ b/exasol/driver/websocket/_types.py
@@ -0,0 +1,91 @@
+"""
+This module provides `PEP-249`_ a DBAPI compliant types and type conversion definitions.
+(see also `PEP-249-types`_)
+
+.. _PEP-249-types: https://peps.python.org/pep-0249/#type-objects-and-constructors
+"""
+
+from datetime import (
+    date,
+    datetime,
+    time,
+)
+from enum import Enum
+from time import localtime
+
+Date = date
+Time = time
+Timestamp = datetime
+
+
+def DateFromTicks(ticks: int) -> date:  # pylint: disable=C0103
+    """
+    This function constructs an object holding a date value from the given ticks value
+    (number of seconds since the epoch; see the documentation of the standard
+    Python time module for details).
+    """
+    year, month, day = localtime(ticks)[:3]
+    return Date(year, month, day)
+
+
+def TimeFromTicks(ticks: int) -> time:  # pylint: disable=C0103
+    """
+    This function constructs an object holding a time value from the given ticks value
+    (number of seconds since the epoch; see the documentation of the standard
+    Python time module for details).
+    """
+    hour, minute, second = localtime(ticks)[3:6]
+    return Time(hour, minute, second)
+
+
+def TimestampFromTicks(ticks: int) -> datetime:  # pylint: disable=C0103
+    """
+    This function constructs an object holding a time stamp value from the
+    given ticks value (number of seconds since the epoch; see the documentation
+    of the standard Python time module for details).
+    """
+    year, month, day, hour, minute, second = localtime(ticks)[:6]
+    return Timestamp(year, month, day, hour, minute, second)
+
+
+class TypeCode(Enum):
+    """
+    Type codes for Exasol DB column types.
+
+    See: https://github.com/exasol/websocket-api/blob/master/docs/WebsocketAPIV3.md#data-types-type-names-and-properties
+    """
+
+    Bool = "BOOLEAN"
+    Char = "CHAR"
+    Date = "DATE"
+    Decimal = "DECIMAL"
+    Double = "DOUBLE"
+    Geometry = "GEOMETRY"
+    IntervalDayToSecond = "INTERVAL DAY TO SECOND"
+    IntervalYearToMonth = "INTERVAL YEAR TO MONTH"
+    Timestamp = "TIMESTAMP"
+    TimestampTz = "TIMESTAMP WITH LOCAL TIME ZONE"
+    String = "VARCHAR"
+
+
+class _DBAPITypeObject:
+    def __init__(self, *type_codes) -> None:
+        self.type_codes = type_codes
+
+    def __eq__(self, other):
+        return other in self.type_codes
+
+
+STRING = _DBAPITypeObject(TypeCode.String)
+# A binary type is not natively supported by Exasol
+BINARY = _DBAPITypeObject(None)
+NUMBER = _DBAPITypeObject(TypeCode.Decimal, TypeCode.Double)
+DATETIME = _DBAPITypeObject(
+    TypeCode.Date,
+    TypeCode.Timestamp,
+    TypeCode.TimestampTz,
+    TypeCode.IntervalDayToSecond,
+    TypeCode.IntervalYearToMonth,
+)
+# Exasol does manage indexes internally
+ROWID = _DBAPITypeObject(None)
diff --git a/exasol/driver/websocket/dbapi2.py b/exasol/driver/websocket/dbapi2.py
new file mode 100644
index 0000000..81f5055
--- /dev/null
+++ b/exasol/driver/websocket/dbapi2.py
@@ -0,0 +1,102 @@
+"""
+This module provides a `PEP-249`_ compliant DBAPI interface, for a websocket based
+database driver (see also `exasol-websocket-api`_).
+
+.. _PEP-249: https://peps.python.org/pep-0249/#interfaceerror
+.. _exasol-websocket-api: https://github.com/exasol/websocket-api
+"""
+
+from exasol.driver.websocket._connection import Connection as DefaultConnection
+
+# Re-export types and definitions required by dbapi2
+from exasol.driver.websocket._errors import (
+    DatabaseError,
+    DataError,
+    Error,
+    IntegrityError,
+    InterfaceError,
+    InternalError,
+    NotSupportedError,
+    OperationalError,
+    ProgrammingError,
+    Warning,
+)
+from exasol.driver.websocket._protocols import (
+    Connection,
+    Cursor,
+)
+from exasol.driver.websocket._types import (
+    BINARY,
+    DATETIME,
+    NUMBER,
+    ROWID,
+    STRING,
+    Date,
+    DateFromTicks,
+    Time,
+    TimeFromTicks,
+    Timestamp,
+    TimestampFromTicks,
+    TypeCode,
+)
+
+# Add remaining definitions
+
+apilevel = "2.0"  # Required by spec. pylint: disable=C0103
+threadsafety = 1  # Required by spec. pylint: disable=C0103
+paramstyle = "qmark"  # Required by spec. pylint: disable=C0103
+
+
+def connect(connection_class=DefaultConnection, **kwargs) -> Connection:
+    """
+    Creates a connection to the database.
+
+    Args:
+        connection_class: which shall be used to construct a connection object.
+        kwargs: compatible with the provided connection_class.
+
+    Returns:
+
+        returns a dbapi2 complaint connection object.
+    """
+    connection = connection_class(**kwargs)
+    return connection.connect()
+
+
+__all__ = [
+    # ----- Constants -----
+    "apilevel",
+    "threadsafety",
+    "paramstyle",
+    # ----- Errors -----
+    "Warning",
+    "Error",
+    "InterfaceError",
+    "DatabaseError",
+    "DataError",
+    "OperationalError",
+    "InternalError",
+    "IntegrityError",
+    "InternalError",
+    "ProgrammingError",
+    "NotSupportedError",
+    # ----- Protocols -----
+    "Connection",
+    "Cursor",
+    # ----- Types and Type-Conversions -----
+    "Date",
+    "Time",
+    "Timestamp",
+    "DateFromTicks",
+    "TimeFromTicks",
+    "TimestampFromTicks",
+    "STRING",
+    "BINARY",
+    "NUMBER",
+    "DATETIME",
+    "ROWID",
+    # ----- Functions ------
+    "connect",
+    # ----- Non DBAPI exports -----
+    "TypeCode",
+]
diff --git a/pyexasol/db2/__init__.py b/pyexasol/db2/__init__.py
index 07718fc..307e17f 100644
--- a/pyexasol/db2/__init__.py
+++ b/pyexasol/db2/__init__.py
@@ -8,16 +8,29 @@
 There is no "paramstyle" and no proper error handling
 """
 
+from warnings import warn
+from inspect import cleandoc
 from ..connection import ExaConnection
-
-apilevel = '2.0'
+from ..warnings import PyexasolDeprecationWarning
+
+warn(
+    cleandoc(
+        """
+        This module is deprecated and will be removed in the future.
+        If you require a dbapi2 compliant driver, please use the dbapi2 compliant driver facade available in the `exasol.driver.websocket` package.
+        """
+    ),
+    PyexasolDeprecationWarning,
+)
+
+apilevel = "2.0"
 threadsafety = 1
 paramstyle = None
 
 
 def connect(**kwargs):
-    if 'autocommit' not in kwargs:
-        kwargs['autocommit'] = False
+    if "autocommit" not in kwargs:
+        kwargs["autocommit"] = False
 
     return DB2Connection(**kwargs)
 
@@ -66,15 +79,17 @@ def description(self):
         cols = []
 
         for k, v in self.stmt.columns().items():
-            cols.append((
-                k,
-                v.get('type', None),
-                v.get('size', None),
-                v.get('size', None),
-                v.get('precision', None),
-                v.get('scale', None),
-                True
-            ))
+            cols.append(
+                (
+                    k,
+                    v.get("type", None),
+                    v.get("size", None),
+                    v.get("size", None),
+                    v.get("precision", None),
+                    v.get("scale", None),
+                    True,
+                )
+            )
 
         return cols
 
diff --git a/pyexasol/warnings.py b/pyexasol/warnings.py
new file mode 100644
index 0000000..0c13b95
--- /dev/null
+++ b/pyexasol/warnings.py
@@ -0,0 +1,6 @@
+class PyexasolWarning(UserWarning):
+    """Base class for all warnings emitted by pyexasol."""
+
+
+class PyexasolDeprecationWarning(PyexasolWarning, DeprecationWarning):
+    """Warning class for features that will be removed in future versions."""
diff --git a/pyproject.toml b/pyproject.toml
index b048ccd..f9f23b7 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -16,6 +16,11 @@ authors = [
 classifiers = [
     "Development Status :: 5 - Production/Stable",
     "License :: OSI Approved :: MIT License",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
     "Intended Audience :: Developers",
     "Topic :: Database",
 ]
diff --git a/test/integration/exasol/conftest.py b/test/integration/exasol/conftest.py
new file mode 100644
index 0000000..e69de29
diff --git a/test/integration/exasol/dbapi/conftest.py b/test/integration/exasol/dbapi/conftest.py
new file mode 100644
index 0000000..d5fb6ff
--- /dev/null
+++ b/test/integration/exasol/dbapi/conftest.py
@@ -0,0 +1,24 @@
+import pytest
+
+from exasol.driver.websocket.dbapi2 import connect
+
+
+@pytest.fixture
+def connection(dsn, user, password, schema):
+    _connection = connect(
+        dsn=dsn,
+        username=user,
+        password=password,
+        schema=schema,
+        certificate_validation=False,
+    )
+    yield _connection
+    _connection.close()
+
+@pytest.fixture
+def cursor(connection):
+    cursor = connection.cursor()
+    yield cursor
+    cursor.close()
+
+
diff --git a/test/integration/exasol/dbapi/dbapi_test.py b/test/integration/exasol/dbapi/dbapi_test.py
new file mode 100644
index 0000000..7ade0e9
--- /dev/null
+++ b/test/integration/exasol/dbapi/dbapi_test.py
@@ -0,0 +1,322 @@
+from inspect import cleandoc
+
+import pytest
+
+from exasol.driver.websocket.dbapi2 import (
+    Error,
+    NotSupportedError,
+    TypeCode,
+    connect,
+)
+
+
+def test_websocket_dbapi(dsn, user, password, schema):
+    connection = connect(
+        dsn=dsn,
+        username=user,
+        password=password,
+        schema=schema,
+        certificate_validation=False,
+    )
+    assert connection
+    connection.close()
+
+
+def test_websocket_dbapi_connect_fails():
+    dsn = "127.0.0.2:9999"
+    username = "ShouldNotExist"
+    password = "ThisShouldNotBeAValidPasswordForTheUser"
+    with pytest.raises(Error) as e_info:
+        connect(dsn=dsn, username=username, password=password)
+    assert "Connection failed" in f"{e_info.value}"
+
+
+def test_retrieve_cursor_from_connection(connection):
+    cursor = connection.cursor()
+    assert cursor
+    cursor.close()
+
+
+@pytest.mark.parametrize(
+    "sql_statement", ["SELECT 1;", "SELECT * FROM VALUES 1, 2, 3, 4;"]
+)
+def test_cursor_execute(cursor, sql_statement):
+    # Because the dbapi does not specify a required return value, this is just a smoke test
+    # to ensure the execute call won't crash.
+    cursor.execute(sql_statement)
+
+
+@pytest.mark.parametrize(
+    "sql_statement, expected",
+    [
+        ("SELECT 1;", (1,)),
+        ("SELECT * FROM VALUES (1, 2, 3);", (1, 2, 3)),
+        ("SELECT * FROM VALUES 1, 5, 9, 13;", (1,)),
+    ],
+    ids=str,
+)
+def test_cursor_fetchone(cursor, sql_statement, expected):
+    cursor.execute(sql_statement)
+    assert cursor.fetchone() == expected
+
+
+@pytest.mark.parametrize("method", ("fetchone", "fetchmany", "fetchall"))
+def test_cursor_function_raises_exception_if_no_result_has_been_produced(
+    cursor, method
+):
+    expected = "No result has been produced."
+    cursor_method = getattr(cursor, method)
+    with pytest.raises(Error) as e_info:
+        cursor_method()
+    assert f"{e_info.value}" == expected
+
+
+@pytest.mark.parametrize(
+    "sql_statement, size, expected",
+    [
+        ("SELECT 1;", None, ((1,),)),
+        ("SELECT 1;", 1, ((1,),)),
+        ("SELECT 1;", 10, ((1,),)),
+        ("SELECT * FROM VALUES ((1,2), (3,4), (5,6));", None, ((1, 2),)),
+        ("SELECT * FROM VALUES ((1,2), (3,4), (5,6));", 1, ((1, 2),)),
+        (
+            "SELECT * FROM VALUES ((1,2), (3,4), (5,6));",
+            2,
+            (
+                (1, 2),
+                (3, 4),
+            ),
+        ),
+        (
+            "SELECT * FROM VALUES ((1,2), (3,4), (5,6));",
+            10,
+            (
+                (1, 2),
+                (3, 4),
+                (5, 6),
+            ),
+        ),
+    ],
+    ids=str,
+)
+def test_cursor_fetchmany(cursor, sql_statement, size, expected):
+    cursor.execute(sql_statement)
+    assert cursor.fetchmany(size) == expected
+
+
+@pytest.mark.parametrize(
+    "sql_statement, expected",
+    [
+        ("SELECT 1;", ((1,),)),
+        (
+            "SELECT * FROM VALUES ((1,2), (3,4));",
+            (
+                (1, 2),
+                (3, 4),
+            ),
+        ),
+        (
+            "SELECT * FROM VALUES ((1,2), (3,4), (5,6));",
+            (
+                (1, 2),
+                (3, 4),
+                (5, 6),
+            ),
+        ),
+        (
+            "SELECT * FROM VALUES ((1,2), (3,4), (5,6), (7, 8));",
+            (
+                (1, 2),
+                (3, 4),
+                (5, 6),
+                (7, 8),
+            ),
+        ),
+    ],
+    ids=str,
+)
+def test_cursor_fetchall(cursor, sql_statement, expected):
+    cursor.execute(sql_statement)
+    assert cursor.fetchall() == expected
+
+
+def test_description_returns_none_if_no_query_has_been_executed(cursor):
+    assert cursor.description is None
+
+
+@pytest.mark.parametrize(
+    "sql_statement, expected",
+    [
+        (
+            "SELECT CAST(A as INT) A FROM VALUES 1, 2, 3 as T(A);",
+            (("A", TypeCode.Decimal, None, None, 18, 0, None),),
+        ),
+        (
+            "SELECT CAST(A as DOUBLE) A FROM VALUES 1, 2, 3 as T(A);",
+            (("A", TypeCode.Double, None, None, None, None, None),),
+        ),
+        (
+            "SELECT CAST(A as BOOL) A FROM VALUES TRUE, FALSE, TRUE as T(A);",
+            (("A", TypeCode.Bool, None, None, None, None, None),),
+        ),
+        (
+            "SELECT CAST(A as VARCHAR(10)) A FROM VALUES 'Foo', 'Bar' as T(A);",
+            (("A", TypeCode.String, None, 10, None, None, None),),
+        ),
+        (
+            cleandoc(
+                # fmt: off
+                    """
+                    SELECT CAST(A as INT) A, CAST(B as VARCHAR(100)) B, CAST(C as BOOL) C, CAST(D as DOUBLE) D
+                    FROM VALUES ((1,'Some String', TRUE, 1.0), (3,'Other String', FALSE, 2.0)) as TB(A, B, C, D);
+                    """
+                # fmt: on
+            ),
+            (
+                ("A", TypeCode.Decimal, None, None, 18, 0, None),
+                ("B", TypeCode.String, None, 100, None, None, None),
+                ("C", TypeCode.Bool, None, None, None, None, None),
+                ("D", TypeCode.Double, None, None, None, None, None),
+            ),
+        ),
+    ],
+    ids=str,
+)
+def test_description_attribute(cursor, sql_statement, expected):
+    cursor.execute(sql_statement)
+    assert cursor.description == expected
+
+
+@pytest.mark.parametrize(
+    "sql_statement,expected",
+    (
+        ("SELECT 1;", 1),
+        ("SELECT * FROM VALUES TRUE, FALSE as T(A);", 2),
+        ("SELECT * FROM VALUES TRUE, FALSE, TRUE as T(A);", 3),
+        # ATTENTION: As of today 03.02.2023 it seems there is no trivial way to make this test pass.
+        #            Also, it is unclear if this semantic is required in order to function correctly
+        #            with SQLA.
+        #
+        #            NOTE: In order to implement this semantic, subclassing pyexasol.ExaConnection and
+        #                  pyexasol.ExaStatement most likely will be required.
+        pytest.param("DROP SCHEMA IF EXISTS FOOBAR;", -1, marks=pytest.mark.xfail),
+    ),
+)
+def test_rowcount_attribute(cursor, sql_statement, expected):
+    cursor.execute(sql_statement)
+    assert cursor.rowcount == expected
+
+
+def test_rowcount_attribute_returns_minus_one_if_no_statement_was_executed_yet(cursor):
+    expected = -1
+    assert cursor.rowcount == expected
+
+
+def test_callproc_is_not_supported(cursor):
+    expected = "Optional and therefore not supported"
+    with pytest.raises(NotSupportedError) as exec_info:
+        cursor.callproc(None)
+    assert f"{exec_info.value}" == expected
+
+
+def test_cursor_nextset_is_not_supported(cursor):
+    expected = "Optional and therefore not supported"
+    with pytest.raises(NotSupportedError) as exec_info:
+        cursor.nextset()
+    assert f"{exec_info.value}" == expected
+
+
+@pytest.mark.parametrize("property", ("arraysize", "description", "rowcount"))
+def test_cursor_closed_cursor_raises_exception_on_property_access(connection, property):
+    expected = (
+        f"Unable to execute operation <{property}>, because cursor was already closed."
+    )
+
+    cursor = connection.cursor()
+    cursor.close()
+
+    with pytest.raises(Error) as exec_info:
+        _ = getattr(cursor, property)
+
+    assert f"{exec_info.value}" == expected
+
+
+@pytest.mark.parametrize(
+    "method,args",
+    (
+        ("callproc", [None]),
+        ("execute", ["SELECT 1;"]),
+        ("executemany", ["SELECT 1;", []]),
+        ("fetchone", []),
+        ("fetchmany", []),
+        ("fetchall", []),
+        ("nextset", []),
+        ("setinputsizes", [None]),
+        ("setoutputsize", [None, None]),
+        ("close", []),
+    ),
+    ids=str,
+)
+def test_cursor_closed_cursor_raises_exception_on_method_usage(
+    connection, method, args
+):
+    expected = (
+        f"Unable to execute operation <{method}>, because cursor was already closed."
+    )
+
+    cursor = connection.cursor()
+    cursor.execute("SELECT 1;")
+    cursor.close()
+
+    with pytest.raises(Error) as exec_info:
+        method = getattr(cursor, method)
+        method(*args)
+
+    assert f"{exec_info.value}" == expected
+
+
+@pytest.fixture
+def test_schema(cursor):
+    schema = "TEST"
+    cursor.execute(f"DROP SCHEMA IF EXISTS {schema} CASCADE")
+    cursor.execute(f"CREATE SCHEMA {schema};")
+    yield schema
+    cursor.execute(f"DROP SCHEMA IF EXISTS {schema} CASCADE")
+
+
+@pytest.fixture
+def users_table(cursor, test_schema):
+    table = "USERS"
+    cursor.execute(f"DROP TABLE IF EXISTS {test_schema}.{table}")
+    cursor.execute(
+        # fmt: off
+        cleandoc(
+            f"""
+            CREATE TABLE {test_schema}.{table} (
+                firstname VARCHAR(100) ,
+                lastname VARCHAR(100),
+                id DECIMAL
+            );
+            """
+        )
+        # fmt: on
+    )
+    yield f"{test_schema}.{table}"
+    cursor.execute(f"DROP TABLE IF EXISTS {test_schema}.{table}")
+
+
+def test_cursor_executemany(users_table, cursor):
+    values = [("John", "Doe", 0), ("Donald", "Duck", 1)]
+
+    cursor.execute(f"SELECT * FROM {users_table};")
+    before = cursor.fetchall()
+
+    cursor.executemany(f"INSERT INTO {users_table} VALUES (?, ?, ?);", values)
+
+    cursor.execute(f"SELECT * FROM {users_table};")
+    after = cursor.fetchall()
+
+    expected = len(values)
+    actual = len(after) - len(before)
+
+    assert actual == expected
diff --git a/test/unit/deprectation_warning_test.py b/test/unit/deprectation_warning_test.py
new file mode 100644
index 0000000..0e06099
--- /dev/null
+++ b/test/unit/deprectation_warning_test.py
@@ -0,0 +1,7 @@
+import pytest
+from pyexasol.warnings import PyexasolDeprecationWarning
+
+
+def test_import_db2_module_emits_deprecation_warning():
+    with pytest.warns(PyexasolDeprecationWarning):
+        from pyexasol import db2
diff --git a/test/unit/exasol/test_dbapi2.py b/test/unit/exasol/test_dbapi2.py
new file mode 100644
index 0000000..a4fb0b6
--- /dev/null
+++ b/test/unit/exasol/test_dbapi2.py
@@ -0,0 +1,148 @@
+"""
+This module contains compatibility tests for pythons dbapi module interface
+"""
+
+import datetime
+import importlib
+
+import pytest
+
+from exasol.driver.websocket._connection import _requires_connection
+from exasol.driver.websocket._cursor import (
+    MetaData,
+    _pyexasol2dbapi_metadata,
+)
+from exasol.driver.websocket.dbapi2 import (
+    Error,
+    TypeCode,
+)
+
+
+@pytest.fixture
+def dbapi():
+    yield importlib.import_module("exasol.driver.websocket.dbapi2")
+
+
+def test_defines_api_level(dbapi):
+    assert dbapi.apilevel in {"1.0", "2.0"}
+
+
+def test_defines_threadsafety(dbapi):
+    assert dbapi.threadsafety in {0, 1, 2, 3}
+
+
+def test_defines_paramstyle(dbapi):
+    assert dbapi.paramstyle in {"qmark", "numeric", "named", "format", "pyformat"}
+
+
+@pytest.mark.parametrize(
+    "exception",
+    [
+        "Warning",
+        "Error",
+        "InterfaceError",
+        "DatabaseError",
+        "DataError",
+        "OperationalError",
+        "IntegrityError",
+        "InternalError",
+        "ProgrammingError",
+        "NotSupportedError",
+    ],
+)
+def test_all_exceptions_are_available(dbapi, exception):
+    assert issubclass(getattr(dbapi, exception), Exception)
+
+
+@pytest.mark.parametrize("year,month,day", [(2022, 12, 24), (2023, 1, 1)])
+def test_date_constructor(dbapi, year, month, day):
+    actual = dbapi.Date(year, month, day)
+    expected = datetime.date(year, month, day)
+    assert actual == expected
+
+
+@pytest.mark.parametrize("hour,minute,second", [(12, 1, 24), (23, 1, 1)])
+def test_time_constructor(dbapi, hour, minute, second):
+    actual = dbapi.Time(hour, minute, second)
+    expected = datetime.time(hour, minute, second)
+    assert actual == expected
+
+
+@pytest.mark.parametrize(
+    "year,month,day,hour,minute,second",
+    [(2022, 12, 24, 12, 1, 24), (2023, 1, 1, 23, 1, 1)],
+)
+def test_timestamp_constructor(dbapi, year, month, day, hour, minute, second):
+    actual = dbapi.Timestamp(year, month, day, hour, minute, second)
+    expected = datetime.datetime(year, month, day, hour, minute, second)
+    assert actual == expected
+
+
+def test_requires_connection_decorator_throws_exception_if_no_connection_is_available():
+    class MyConnection:
+        def __init__(self, con=None):
+            self._connection = con
+
+        @_requires_connection
+        def close(self):
+            pass
+
+        def connect(self):
+            self._connection = object()
+
+    connection = MyConnection()
+    with pytest.raises(Error) as e_info:
+        connection.close()
+
+    assert "No active connection available" == f"{e_info.value}"
+
+
+def test_requires_connection_decorator_does_not_throw_exception_connection_is_available():
+    class MyConnection:
+        def __init__(self, con=None):
+            self._connection = con
+
+        @_requires_connection
+        def close(self):
+            return self._connection
+
+        def connect(self):
+            self._connection = object()
+
+    connection = MyConnection(con=object())
+    assert connection.close()
+
+
+def test_requires_connection_decorator_does_use_wrap():
+    class MyConnection:
+        @_requires_connection
+        def close(self):
+            return True
+
+    connection = MyConnection()
+    assert "close" == connection.close.__name__
+
+
+@pytest.mark.parametrize(
+    "name,metadata,expected",
+    (
+        (
+            (
+                "A",
+                {"type": "DECIMAL", "precision": 18, "scale": 0},
+                MetaData(name="A", type_code=TypeCode.Decimal, precision=18, scale=0),
+            ),
+            (
+                "B",
+                {"type": "VARCHAR", "size": 100, "characterSet": "UTF8"},
+                MetaData(name="B", type_code=TypeCode.String, internal_size=100),
+            ),
+            ("C", {"type": "BOOLEAN"}, MetaData(name="C", type_code=TypeCode.Bool)),
+            ("D", {"type": "DOUBLE"}, MetaData(name="D", type_code=TypeCode.Double)),
+        )
+    ),
+    ids=str,
+)
+def test_metadata_from_pyexasol_metadata(name, metadata, expected):
+    actual = _pyexasol2dbapi_metadata(name, metadata)
+    assert actual == expected