Repository: trinodb/trino-python-client

Issue:
USE catalog.schema not honored
The connection can be created with a catalog and schema argument. However I would have expected that the SQL USE statement would also switch between catalogs and schemas. Here's an example
```
import trino
conn = trino.dbapi.connect(
                host="localhost",
                port=8080,
                auth=None,
                user='the-user',
                http_scheme="http")
cur = conn.cursor()

# this works fine since table is fully qualified
cur.execute("select * from tpch.tiny.orders")
results = cur.fetchmany(10)
print(results)

# The USE statement to switch schema
cur.execute("use tpch.tiny")

# Query will fail because the USE statement is not honored
cur.execute("select * from orders")
results = cur.fetchmany(10)
print(results)
```


Discussion:
mdesmet: Not sure why the build failed in 3.10 on the following error:

```
trino.exceptions.TrinoUserError: TrinoUserError(type=USER_ERROR, name=COLUMN_NOT_FOUND, message="line 1:8: Column 'current_catalog' cannot be resolved", query_id=20220424_155121_00213_82y8r)

```
ebyhr: `current_catalog` was added in version 357, so it fails on 351.
https://trino.io/docs/current/release/release-357.html
mdesmet: I have rebased the code and made following changes:

- `ClientSession` is now thread-safe. it is initiated by the `dbapi.Connection` and some of its properties can be modified based on the headers coming back from the Trino server. It can later on be extended with more modifiable properties. 
- Added tests for both `USE catalog.schema` and `USE schema`
- Improved mypy typing for `ClientSession` class

Patch:
diff --git a/README.md b/README.md
index 3ea1264..a0ae128 100644
--- a/README.md
+++ b/README.md
@@ -99,7 +99,8 @@ engine = create_engine(
     'trino://user@localhost:8080/system',
     connect_args={
       "session_properties": {'query_max_run_time': '1d'},
-      "client_tags": ["tag1", "tag2"]
+      "client_tags": ["tag1", "tag2"],
+      "experimental_python_types": True,
     }
 )
 
@@ -107,7 +108,8 @@ engine = create_engine(
 engine = create_engine(
     'trino://user@localhost:8080/system?'
     'session_properties={"query_max_run_time": "1d"}'
-    '&client_tags=["tag1", "tag2"]',
+    '&client_tags=["tag1", "tag2"]'
+    '&experimental_python_types=true',
 )
 ```
 
@@ -361,10 +363,11 @@ import pytz
 from datetime import datetime
 
 conn = trino.dbapi.connect(
+    experimental_python_types=True,
     ...
 )
 
-cur = conn.cursor(experimental_python_types=True)
+cur = conn.cursor()
 
 params = datetime(2020, 1, 1, 16, 43, 22, 320000, tzinfo=pytz.timezone('America/Los_Angeles'))
 
diff --git a/trino/client.py b/trino/client.py
index 1362975..3e19859 100644
--- a/trino/client.py
+++ b/trino/client.py
@@ -36,6 +36,7 @@ The main interface is :class:`TrinoQuery`: ::
 import copy
 import os
 import re
+import threading
 import urllib.parse
 from datetime import datetime, timedelta, timezone
 from decimal import Decimal
@@ -46,9 +47,8 @@ import requests
 
 import trino.logging
 from trino import constants, exceptions
-from trino.transaction import NO_TRANSACTION
 
-__all__ = ["TrinoQuery", "TrinoRequest", "PROXIES"]
+__all__ = ["ClientSession", "TrinoQuery", "TrinoRequest", "PROXIES"]
 
 logger = trino.logging.get_logger(__name__)
 
@@ -67,38 +67,128 @@ NAN = float("nan")
 
 
 class ClientSession(object):
+    """
+    Manage the current Client Session properties of a specific connection. This class is thread-safe.
+
+    :param user: associated with the query. It is useful for access control
+                 and query scheduling.
+    :param source: associated with the query. It is useful for access
+                   control and query scheduling.
+    :param catalog: to query. The *catalog* is associated with a Trino
+                    connector. This variable sets the default catalog used
+                    by SQL statements. For example, if *catalog* is set
+                    to ``some_catalog``, the SQL statement
+                    ``SELECT * FROM some_schema.some_table`` will actually
+                    query the table
+                    ``some_catalog.some_schema.some_table``.
+    :param schema: to query. The *schema* is a logical abstraction to group
+                   table. This variable sets the default schema used by
+                   SQL statements. For example, if *schema* is set to
+                   ``some_schema``, the SQL statement
+                   ``SELECT * FROM some_table`` will actually query the
+                   table ``some_catalog.some_schema.some_table``.
+    :param properties: set specific Trino behavior for the current
+                               session. Please refer to the output of
+                               ``SHOW SESSION`` to check the available
+                               properties.
+    :param headers: HTTP headers to POST/GET in the HTTP requests
+    :param extra_credential: extra credentials. as list of ``(key, value)``
+                             tuples.
+    :param client_tags: Client tags as list of strings.
+    """
+
     def __init__(
         self,
-        catalog,
-        schema,
-        source,
-        user,
-        properties=None,
-        headers=None,
-        transaction_id=None,
-        extra_credential=None,
-        client_tags=None
+        user: str,
+        catalog: str = None,
+        schema: str = None,
+        source: str = None,
+        properties: Dict[str, str] = None,
+        headers: Dict[str, str] = None,
+        transaction_id: str = None,
+        extra_credential: List[Tuple[str, str]] = None,
+        client_tags: List[str] = None,
     ):
-        self.catalog = catalog
-        self.schema = schema
-        self.source = source
-        self.user = user
-        if properties is None:
-            properties = {}
-        self._properties = properties
+        self._user = user
+        self._catalog = catalog
+        self._schema = schema
+        self._source = source
+        self._properties = properties or {}
         self._headers = headers or {}
-        self.transaction_id = transaction_id
-        self.extra_credential = extra_credential
-        self.client_tags = client_tags
+        self._transaction_id = transaction_id
+        self._extra_credential = extra_credential
+        self._client_tags = client_tags
+        self._object_lock = threading.Lock()
+
+    @property
+    def user(self):
+        return self._user
+
+    @property
+    def catalog(self):
+        with self._object_lock:
+            return self._catalog
+
+    @catalog.setter
+    def catalog(self, catalog):
+        with self._object_lock:
+            self._catalog = catalog
+
+    @property
+    def schema(self):
+        with self._object_lock:
+            return self._schema
+
+    @schema.setter
+    def schema(self, schema):
+        with self._object_lock:
+            self._schema = schema
+
+    @property
+    def source(self):
+        return self._source
 
     @property
     def properties(self):
-        return self._properties
+        with self._object_lock:
+            return self._properties
+
+    @properties.setter
+    def properties(self, properties):
+        with self._object_lock:
+            self._properties = properties
 
     @property
     def headers(self):
         return self._headers
 
+    @property
+    def transaction_id(self):
+        with self._object_lock:
+            return self._transaction_id
+
+    @transaction_id.setter
+    def transaction_id(self, transaction_id):
+        with self._object_lock:
+            self._transaction_id = transaction_id
+
+    @property
+    def extra_credential(self):
+        return self._extra_credential
+
+    @property
+    def client_tags(self):
+        return self._client_tags
+
+    def __getstate__(self):
+        state = self.__dict__.copy()
+        del state["_object_lock"]
+        return state
+
+    def __setstate__(self, state):
+        self.__dict__.update(state)
+        self._object_lock = threading.Lock()
+
 
 def get_header_values(headers, header):
     return [val.strip() for val in headers[header].split(",")]
@@ -143,33 +233,9 @@ class TrinoRequest(object):
 
     :param host: name of the coordinator
     :param port: TCP port to connect to the coordinator
-    :param user: associated with the query. It is useful for access control
-                 and query scheduling.
-    :param source: associated with the query. It is useful for access
-                   control and query scheduling.
-    :param catalog: to query. The *catalog* is associated with a Trino
-                    connector. This variable sets the default catalog used
-                    by SQL statements. For example, if *catalog* is set
-                    to ``some_catalog``, the SQL statement
-                    ``SELECT * FROM some_schema.some_table`` will actually
-                    query the table
-                    ``some_catalog.some_schema.some_table``.
-    :param schema: to query. The *schema* is a logical abstraction to group
-                   table. This variable sets the default schema used by
-                   SQL statements. For eample, if *schema* is set to
-                   ``some_schema``, the SQL statement
-                   ``SELECT * FROM some_table`` will actually query the
-                   table ``some_catalog.some_schema.some_table``.
-    :param session_properties: set specific Trino behavior for the current
-                               session. Please refer to the output of
-                               ``SHOW SESSION`` to check the available
-                               properties.
-    :param http_headers: HTTP headers to post/get in the HTTP requests
     :param http_scheme: "http" or "https"
     :param auth: class that manages user authentication. ``None`` means no
                  authentication.
-    :param extra_credential: extra credentials. as list of ``(key, value)``
-                             tuples.
     :max_attempts: maximum number of attempts when sending HTTP requests. An
                    attempt is an HTTP request. 5 attempts means 4 retries.
     :request_timeout: How long (in seconds) to wait for the server to send
@@ -213,36 +279,17 @@ class TrinoRequest(object):
         self,
         host: str,
         port: int,
-        user: str,
-        source: str = None,
-        catalog: str = None,
-        schema: str = None,
-        session_properties: Optional[Dict[str, Any]] = None,
+        client_session: ClientSession,
         http_session: Any = None,
-        http_headers: Optional[Dict[str, str]] = None,
-        transaction_id: Optional[str] = NO_TRANSACTION,
         http_scheme: str = None,
         auth: Optional[Any] = constants.DEFAULT_AUTH,
-        extra_credential: Optional[List[Tuple[str, str]]] = None,
         redirect_handler: Any = None,
         max_attempts: int = MAX_ATTEMPTS,
         request_timeout: Union[float, Tuple[float, float]] = constants.DEFAULT_REQUEST_TIMEOUT,
         handle_retry=exceptions.RetryWithExponentialBackoff(),
         verify: bool = True,
-        client_tags: Optional[List[str]] = None
     ) -> None:
-        self._client_session = ClientSession(
-            catalog,
-            schema,
-            source,
-            user,
-            session_properties,
-            http_headers,
-            transaction_id,
-            extra_credential,
-            client_tags
-        )
-
+        self._client_session = client_session
         self._host = host
         self._port = port
         self._next_uri: Optional[str] = None
@@ -454,6 +501,12 @@ class TrinoRequest(object):
             ):
                 self._client_session.properties[key] = value
 
+        if constants.HEADER_SET_CATALOG in http_response.headers:
+            self._client_session.catalog = http_response.headers[constants.HEADER_SET_CATALOG]
+
+        if constants.HEADER_SET_SCHEMA in http_response.headers:
+            self._client_session.schema = http_response.headers[constants.HEADER_SET_SCHEMA]
+
         self._next_uri = response.get("nextUri")
 
         return TrinoStatus(
@@ -504,7 +557,7 @@ class TrinoResult(object):
         # Initial fetch from the first POST request
         for row in self._rows:
             self._rownumber += 1
-            yield row
+            yield self._map_row(self._experimental_python_types, row, self._query.columns)
         self._rows = None
 
         # Subsequent fetches from GET requests until next_uri is empty.
@@ -513,15 +566,19 @@ class TrinoResult(object):
             for row in rows:
                 self._rownumber += 1
                 logger.debug("row %s", row)
-                if not self._experimental_python_types:
-                    yield row
-                else:
-                    yield self._map_to_python_types(row, self._query.columns)
+                yield self._map_row(self._experimental_python_types, row, self._query.columns)
 
     @property
     def response_headers(self):
         return self._query.response_headers
 
+    @classmethod
+    def _map_row(cls, experimental_python_types, row, columns):
+        if not experimental_python_types:
+            return row
+        else:
+            return cls._map_to_python_types(cls, row, columns)
+
     @classmethod
     def _map_to_python_type(cls, item: Tuple[Any, Dict]) -> Any:
         (value, data_type) = item
diff --git a/trino/constants.py b/trino/constants.py
index 5c0e4ab..d225fe3 100644
--- a/trino/constants.py
+++ b/trino/constants.py
@@ -45,3 +45,6 @@ HEADER_TRANSACTION = "X-Trino-Transaction-Id"
 HEADER_PREPARED_STATEMENT = 'X-Trino-Prepared-Statement'
 HEADER_ADDED_PREPARE = 'X-Trino-Added-Prepare'
 HEADER_DEALLOCATED_PREPARE = 'X-Trino-Deallocated-Prepare'
+
+HEADER_SET_SCHEMA = "X-Trino-Set-Schema"
+HEADER_SET_CATALOG = "X-Trino-Set-Catalog"
diff --git a/trino/dbapi.py b/trino/dbapi.py
index 5353fb7..4481316 100644
--- a/trino/dbapi.py
+++ b/trino/dbapi.py
@@ -109,7 +109,8 @@ class Connection(object):
         isolation_level=IsolationLevel.AUTOCOMMIT,
         verify=True,
         http_session=None,
-        client_tags=None
+        client_tags=None,
+        experimental_python_types=False,
     ):
         self.host = host
         self.port = port
@@ -118,6 +119,17 @@ class Connection(object):
         self.catalog = catalog
         self.schema = schema
         self.session_properties = session_properties
+        self._client_session = trino.client.ClientSession(
+            user=user,
+            catalog=catalog,
+            schema=schema,
+            source=source,
+            properties=session_properties,
+            headers=http_headers,
+            transaction_id=NO_TRANSACTION,
+            extra_credential=extra_credential,
+            client_tags=client_tags
+        )
         # mypy cannot follow module import
         if http_session is None:
             self._http_session = trino.client.TrinoRequest.http.Session()
@@ -136,6 +148,7 @@ class Connection(object):
         self._isolation_level = isolation_level
         self._request = None
         self._transaction = None
+        self.experimental_python_types = experimental_python_types
 
     @property
     def isolation_level(self):
@@ -182,34 +195,30 @@ class Connection(object):
         return trino.client.TrinoRequest(
             self.host,
             self.port,
-            self.user,
-            self.source,
-            self.catalog,
-            self.schema,
-            self.session_properties,
+            self._client_session,
             self._http_session,
-            self.http_headers,
-            NO_TRANSACTION,
             self.http_scheme,
             self.auth,
-            self.extra_credential,
             self.redirect_handler,
             self.max_attempts,
             self.request_timeout,
-            client_tags=self.client_tags
         )
 
-    def cursor(self, experimental_python_types=False):
+    def cursor(self, experimental_python_types: bool = None):
         """Return a new :py:class:`Cursor` object using the connection."""
         if self.isolation_level != IsolationLevel.AUTOCOMMIT:
             if self.transaction is None:
                 self.start_transaction()
-            request = self.transaction._request
-        elif self.transaction is not None:
-            request = self.transaction._request
+        if self.transaction is not None:
+            request = self.transaction.request
         else:
             request = self._create_request()
-        return Cursor(self, request, experimental_python_types)
+        return Cursor(
+            self,
+            request,
+            # if experimental_python_types is not explicitly set in Cursor, take from Connection
+            experimental_python_types if experimental_python_types is not None else self.experimental_python_types
+        )
 
 
 class Cursor(object):
diff --git a/trino/sqlalchemy/dialect.py b/trino/sqlalchemy/dialect.py
index 62e493e..50c3a0b 100644
--- a/trino/sqlalchemy/dialect.py
+++ b/trino/sqlalchemy/dialect.py
@@ -122,6 +122,9 @@ class TrinoDialect(DefaultDialect):
         if "client_tags" in url.query:
             kwargs["client_tags"] = json.loads(url.query["client_tags"])
 
+        if "experimental_python_types" in url.query:
+            kwargs["experimental_python_types"] = json.loads(url.query["experimental_python_types"])
+
         return args, kwargs
 
     def get_columns(self, connection: Connection, table_name: str, schema: str = None, **kw) -> List[Dict[str, Any]]:
diff --git a/trino/transaction.py b/trino/transaction.py
index dcb30ac..e6c8523 100644
--- a/trino/transaction.py
+++ b/trino/transaction.py
@@ -59,6 +59,10 @@ class Transaction(object):
     def id(self):
         return self._id
 
+    @property
+    def request(self):
+        return self._request
+
     def begin(self):
         response = self._request.post(START_TRANSACTION)
         if not response.ok:


Test patch:
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index 5869151..6923e95 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -20,7 +20,7 @@ from uuid import uuid4
 import click
 import trino.logging
 import pytest
-from trino.client import TrinoQuery, TrinoRequest
+from trino.client import TrinoQuery, TrinoRequest, ClientSession
 from trino.constants import DEFAULT_PORT
 from trino.exceptions import TimeoutError
 
@@ -65,7 +65,13 @@ def start_trino(image_tag=None):
 
 
 def wait_for_trino_workers(host, port, timeout=180):
-    request = TrinoRequest(host=host, port=port, user="test_fixture")
+    request = TrinoRequest(
+        host=host,
+        port=port,
+        client_session=ClientSession(
+            user="test_fixture"
+        )
+    )
     sql = "SELECT state FROM system.runtime.nodes"
     t0 = time.time()
     while True:
diff --git a/tests/integration/test_dbapi_integration.py b/tests/integration/test_dbapi_integration.py
index ebb6efa..ab5b1f4 100644
--- a/tests/integration/test_dbapi_integration.py
+++ b/tests/integration/test_dbapi_integration.py
@@ -166,8 +166,34 @@ def test_execute_many_select(trino_connection):
     assert "Query must return update type" in str(e.value)
 
 
-def test_python_types_not_used_when_experimental_python_types_is_not_set(trino_connection):
-    cur = trino_connection.cursor()
+@pytest.mark.parametrize("connection_experimental_python_types,cursor_experimental_python_types,expected",
+                         [
+                             (None, None, False),
+                             (None, False, False),
+                             (None, True, True),
+                             (False, None, False),
+                             (False, False, False),
+                             (False, True, True),
+                             (True, None, True),
+                             (True, False, False),
+                             (True, True, True),
+                         ])
+def test_experimental_python_types_with_connection_and_cursor(
+        connection_experimental_python_types,
+        cursor_experimental_python_types,
+        expected,
+        run_trino
+):
+    _, host, port = run_trino
+
+    connection = trino.dbapi.Connection(
+        host=host,
+        port=port,
+        user="test",
+        experimental_python_types=connection_experimental_python_types,
+    )
+
+    cur = connection.cursor(experimental_python_types=cursor_experimental_python_types)
 
     cur.execute("""
     SELECT
@@ -180,15 +206,23 @@ def test_python_types_not_used_when_experimental_python_types_is_not_set(trino_c
     """)
     rows = cur.fetchall()
 
-    for value in rows[0]:
-        assert isinstance(value, str)
+    if expected:
+        assert rows[0][0] == Decimal('0.142857')
+        assert rows[0][1] == date(2018, 1, 1)
+        assert rows[0][2] == datetime(2019, 1, 1, tzinfo=timezone(timedelta(hours=1)))
+        assert rows[0][3] == datetime(2019, 1, 1, tzinfo=pytz.timezone('UTC'))
+        assert rows[0][4] == datetime(2019, 1, 1)
+        assert rows[0][5] == time(0, 0, 0, 0)
+    else:
+        for value in rows[0]:
+            assert isinstance(value, str)
 
-    assert rows[0][0] == '0.142857'
-    assert rows[0][1] == '2018-01-01'
-    assert rows[0][2] == '2019-01-01 00:00:00.000 +01:00'
-    assert rows[0][3] == '2019-01-01 00:00:00.000 UTC'
-    assert rows[0][4] == '2019-01-01 00:00:00.000'
-    assert rows[0][5] == '00:00:00.000'
+        assert rows[0][0] == '0.142857'
+        assert rows[0][1] == '2018-01-01'
+        assert rows[0][2] == '2019-01-01 00:00:00.000 +01:00'
+        assert rows[0][3] == '2019-01-01 00:00:00.000 UTC'
+        assert rows[0][4] == '2019-01-01 00:00:00.000'
+        assert rows[0][5] == '00:00:00.000'
 
 
 def test_decimal_query_param(trino_connection):
@@ -949,3 +983,54 @@ def retrieve_client_tags_from_query(run_trino, client_tags):
 
     query_client_tags = query_info['session']['clientTags']
     return query_client_tags
+
+
+@pytest.mark.skipif(trino_version() == '351', reason="current_catalog not supported in older Trino versions")
+def test_use_catalog_schema(trino_connection):
+    cur = trino_connection.cursor()
+    cur.execute('SELECT current_catalog, current_schema')
+    result = cur.fetchall()
+    assert result[0][0] is None
+    assert result[0][1] is None
+
+    cur.execute('USE tpch.tiny')
+    cur.fetchall()
+    cur.execute('SELECT current_catalog, current_schema')
+    result = cur.fetchall()
+    assert result[0][0] == 'tpch'
+    assert result[0][1] == 'tiny'
+
+    cur.execute('USE tpcds.sf1')
+    cur.fetchall()
+    cur.execute('SELECT current_catalog, current_schema')
+    result = cur.fetchall()
+    assert result[0][0] == 'tpcds'
+    assert result[0][1] == 'sf1'
+
+
+@pytest.mark.skipif(trino_version() == '351', reason="current_catalog not supported in older Trino versions")
+def test_use_catalog(run_trino):
+    _, host, port = run_trino
+
+    trino_connection = trino.dbapi.Connection(
+        host=host, port=port, user="test", source="test", catalog="tpch", max_attempts=1
+    )
+    cur = trino_connection.cursor()
+    cur.execute('SELECT current_catalog, current_schema')
+    result = cur.fetchall()
+    assert result[0][0] == 'tpch'
+    assert result[0][1] is None
+
+    cur.execute('USE tiny')
+    cur.fetchall()
+    cur.execute('SELECT current_catalog, current_schema')
+    result = cur.fetchall()
+    assert result[0][0] == 'tpch'
+    assert result[0][1] == 'tiny'
+
+    cur.execute('USE sf1')
+    cur.fetchall()
+    cur.execute('SELECT current_catalog, current_schema')
+    result = cur.fetchall()
+    assert result[0][0] == 'tpch'
+    assert result[0][1] == 'sf1'
diff --git a/tests/unit/sqlalchemy/test_dialect.py b/tests/unit/sqlalchemy/test_dialect.py
index 861b513..b17f8cf 100644
--- a/tests/unit/sqlalchemy/test_dialect.py
+++ b/tests/unit/sqlalchemy/test_dialect.py
@@ -47,7 +47,8 @@ class TestTrinoDialect:
                     'session_properties={"query_max_run_time": "1d"}'
                     '&http_headers={"trino": 1}'
                     '&extra_credential=[("a", "b"), ("c", "d")]'
-                    '&client_tags=[1, "sql"]'),
+                    '&client_tags=[1, "sql"]'
+                    '&experimental_python_types=true'),
                 list(),
                 dict(
                     host="localhost",
@@ -58,7 +59,8 @@ class TestTrinoDialect:
                     session_properties={"query_max_run_time": "1d"},
                     http_headers={"trino": 1},
                     extra_credential=[("a", "b"), ("c", "d")],
-                    client_tags=[1, "sql"]
+                    client_tags=[1, "sql"],
+                    experimental_python_types=True,
                 ),
             ),
         ],
diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py
index e865310..313a57d 100644
--- a/tests/unit/test_client.py
+++ b/tests/unit/test_client.py
@@ -28,7 +28,7 @@ from tests.unit.oauth_test_utils import RedirectHandler, GetTokenCallback, PostS
     SERVER_ADDRESS
 from trino import constants
 from trino.auth import KerberosAuthentication, _OAuth2TokenBearer
-from trino.client import TrinoQuery, TrinoRequest, TrinoResult
+from trino.client import TrinoQuery, TrinoRequest, TrinoResult, ClientSession
 
 
 @mock.patch("trino.client.TrinoRequest.http")
@@ -38,12 +38,14 @@ def test_trino_initial_request(mock_requests, sample_post_response_data):
     req = TrinoRequest(
         host="coordinator",
         port=8080,
-        user="test",
-        source="test",
-        catalog="test",
-        schema="test",
+        client_session=ClientSession(
+            user="test",
+            source="test",
+            catalog="test",
+            schema="test",
+            properties={},
+        ),
         http_scheme="http",
-        session_properties={},
     )
 
     http_resp = TrinoRequest.http.Response()
@@ -69,16 +71,17 @@ def test_request_headers(mock_get_and_post):
     req = TrinoRequest(
         host="coordinator",
         port=8080,
-        user=user,
-        source=source,
-        catalog=catalog,
-        schema=schema,
+        client_session=ClientSession(
+            user=user,
+            source=source,
+            catalog=catalog,
+            schema=schema,
+            headers={
+                accept_encoding_header: accept_encoding_value,
+                client_info_header: client_info_value,
+            }
+        ),
         http_scheme="http",
-        session_properties={},
-        http_headers={
-            accept_encoding_header: accept_encoding_value,
-            client_info_header: client_info_value,
-        },
         redirect_handler=None,
     )
 
@@ -107,12 +110,14 @@ def test_request_session_properties_headers(mock_get_and_post):
     req = TrinoRequest(
         host="coordinator",
         port=8080,
-        user="test_user",
-        session_properties={
-            "a": "1",
-            "b": "2",
-            "c": "more=v1,v2"
-        }
+        client_session=ClientSession(
+            user="test_user",
+            properties={
+                "a": "1",
+                "b": "2",
+                "c": "more=v1,v2"
+            }
+        )
     )
 
     def assert_headers(headers):
@@ -137,12 +142,14 @@ def test_additional_request_post_headers(mock_get_and_post):
     req = TrinoRequest(
         host="coordinator",
         port=8080,
-        user="test",
-        source="test",
-        catalog="test",
-        schema="test",
+        client_session=ClientSession(
+            user="test",
+            source="test",
+            catalog="test",
+            schema="test",
+            properties={},
+        ),
         http_scheme="http",
-        session_properties={},
     )
 
     sql = 'select 1'
@@ -166,8 +173,10 @@ def test_request_invalid_http_headers():
         TrinoRequest(
             host="coordinator",
             port=8080,
-            user="test",
-            http_headers={constants.HEADER_USER: "invalid_header"},
+            client_session=ClientSession(
+                user="test",
+                headers={constants.HEADER_USER: "invalid_header"},
+            ),
         )
     assert str(value_error.value).startswith("cannot override reserved HTTP header")
 
@@ -178,8 +187,10 @@ def test_request_client_tags_headers(mock_get_and_post):
     req = TrinoRequest(
         host="coordinator",
         port=8080,
-        user="test_user",
-        client_tags=["tag1", "tag2"]
+        client_session=ClientSession(
+            user="test_user",
+            client_tags=["tag1", "tag2"]
+        ),
     )
 
     def assert_headers(headers):
@@ -200,7 +211,9 @@ def test_request_client_tags_headers_no_client_tags(mock_get_and_post):
     req = TrinoRequest(
         host="coordinator",
         port=8080,
-        user="test_user"
+        client_session=ClientSession(
+            user="test_user",
+        )
     )
 
     def assert_headers(headers):
@@ -221,7 +234,9 @@ def test_enabling_https_automatically_when_using_port_443(mock_get_and_post):
     req = TrinoRequest(
         host="coordinator",
         port=constants.DEFAULT_TLS_PORT,
-        user="test",
+        client_session=ClientSession(
+            user="test",
+        ),
     )
 
     req.post("SELECT 1")
@@ -237,7 +252,9 @@ def test_https_scheme(mock_get_and_post):
     req = TrinoRequest(
         host="coordinator",
         port=constants.DEFAULT_TLS_PORT,
-        user="test",
+        client_session=ClientSession(
+            user="test",
+        ),
         http_scheme=constants.HTTPS,
     )
 
@@ -255,7 +272,9 @@ def test_http_scheme_with_port(mock_get_and_post):
     req = TrinoRequest(
         host="coordinator",
         port=constants.DEFAULT_TLS_PORT,
-        user="test",
+        client_session=ClientSession(
+            user="test",
+        ),
         http_scheme=constants.HTTP,
     )
 
@@ -287,7 +306,9 @@ def test_request_timeout():
         req = TrinoRequest(
             host=host,
             port=port,
-            user="test",
+            client_session=ClientSession(
+                user="test",
+            ),
             http_scheme=http_scheme,
             max_attempts=1,
             request_timeout=request_timeout,
@@ -332,7 +353,9 @@ def test_oauth2_authentication_flow(attempts, sample_post_response_data):
     request = TrinoRequest(
         host="coordinator",
         port=constants.DEFAULT_TLS_PORT,
-        user="test",
+        client_session=ClientSession(
+            user="test",
+        ),
         http_scheme=constants.HTTPS,
         auth=trino.auth.OAuth2Authentication(redirect_auth_url_handler=redirect_handler))
     response = request.post("select 1")
@@ -373,7 +396,9 @@ def test_oauth2_exceed_max_attempts(attempts, sample_post_response_data):
     request = TrinoRequest(
         host="coordinator",
         port=constants.DEFAULT_TLS_PORT,
-        user="test",
+        client_session=ClientSession(
+            user="test",
+        ),
         http_scheme=constants.HTTPS,
         auth=trino.auth.OAuth2Authentication(redirect_auth_url_handler=redirect_handler))
     with pytest.raises(trino.exceptions.TrinoAuthError) as exp:
@@ -405,7 +430,9 @@ def test_oauth2_authentication_missing_headers(header, error):
     request = TrinoRequest(
         host="coordinator",
         port=constants.DEFAULT_TLS_PORT,
-        user="test",
+        client_session=ClientSession(
+            user="test",
+        ),
         http_scheme=constants.HTTPS,
         auth=trino.auth.OAuth2Authentication(redirect_auth_url_handler=RedirectHandler()))
 
@@ -456,7 +483,9 @@ def test_oauth2_header_parsing(header, sample_post_response_data):
     response = TrinoRequest(
         host="coordinator",
         port=constants.DEFAULT_TLS_PORT,
-        user="test",
+        client_session=ClientSession(
+            user="test",
+        ),
         http_scheme=constants.HTTPS,
         auth=trino.auth.OAuth2Authentication(redirect_auth_url_handler=redirect_handler)
     ).post("select 1")
@@ -496,7 +525,9 @@ def test_oauth2_authentication_fail_token_server(http_status, sample_post_respon
     request = TrinoRequest(
         host="coordinator",
         port=constants.DEFAULT_TLS_PORT,
-        user="test",
+        client_session=ClientSession(
+            user="test",
+        ),
         http_scheme=constants.HTTPS,
         auth=trino.auth.OAuth2Authentication(redirect_auth_url_handler=redirect_handler))
 
@@ -527,7 +558,9 @@ def test_multithreaded_oauth2_authentication_flow(sample_post_response_data):
             request = TrinoRequest(
                 host="coordinator",
                 port=constants.DEFAULT_TLS_PORT,
-                user="test",
+                client_session=ClientSession(
+                    user="test",
+                ),
                 http_scheme=constants.HTTPS,
                 auth=auth)
             for i in range(10):
@@ -567,12 +600,14 @@ def test_trino_fetch_request(mock_requests, sample_get_response_data):
     req = TrinoRequest(
         host="coordinator",
         port=8080,
-        user="test",
-        source="test",
-        catalog="test",
-        schema="test",
+        client_session=ClientSession(
+            user="test",
+            source="test",
+            catalog="test",
+            schema="test",
+            properties={},
+        ),
         http_scheme="http",
-        session_properties={},
     )
 
     http_resp = TrinoRequest.http.Response()
@@ -591,12 +626,14 @@ def test_trino_fetch_error(mock_requests, sample_get_error_response_data):
     req = TrinoRequest(
         host="coordinator",
         port=8080,
-        user="test",
-        source="test",
-        catalog="test",
-        schema="test",
+        client_session=ClientSession(
+            user="test",
+            source="test",
+            catalog="test",
+            schema="test",
+            properties={},
+        ),
         http_scheme="http",
-        session_properties={},
     )
 
     http_resp = TrinoRequest.http.Response()
@@ -633,12 +670,14 @@ def test_trino_connection_error(monkeypatch, error_code, error_type, error_messa
     req = TrinoRequest(
         host="coordinator",
         port=8080,
-        user="test",
-        source="test",
-        catalog="test",
-        schema="test",
+        client_session=ClientSession(
+            user="test",
+            source="test",
+            catalog="test",
+            schema="test",
+            properties={},
+        ),
         http_scheme="http",
-        session_properties={},
     )
 
     http_resp = TrinoRequest.http.Response()
@@ -654,8 +693,10 @@ def test_extra_credential(mock_get_and_post):
     req = TrinoRequest(
         host="coordinator",
         port=constants.DEFAULT_TLS_PORT,
-        user="test",
-        extra_credential=[("a.username", "foo"), ("b.password", "bar")],
+        client_session=ClientSession(
+            user="test",
+            extra_credential=[("a.username", "foo"), ("b.password", "bar")],
+        ),
     )
 
     req.post("SELECT 1")
@@ -670,8 +711,10 @@ def test_extra_credential_key_with_illegal_chars():
         TrinoRequest(
             host="coordinator",
             port=constants.DEFAULT_TLS_PORT,
-            user="test",
-            extra_credential=[("a=b", "")],
+            client_session=ClientSession(
+                user="test",
+                extra_credential=[("a=b", "")],
+            ),
         )
 
     assert str(e_info.value) == "whitespace or '=' are disallowed in extra credential 'a=b'"
@@ -682,8 +725,10 @@ def test_extra_credential_key_non_ascii():
         TrinoRequest(
             host="coordinator",
             port=constants.DEFAULT_TLS_PORT,
-            user="test",
-            extra_credential=[("的", "")],
+            client_session=ClientSession(
+                user="test",
+                extra_credential=[("的", "")],
+            ),
         )
 
     assert str(e_info.value) == "only ASCII characters are allowed in extra credential '的'"
@@ -695,8 +740,10 @@ def test_extra_credential_value_encoding(mock_get_and_post):
     req = TrinoRequest(
         host="coordinator",
         port=constants.DEFAULT_TLS_PORT,
-        user="test",
-        extra_credential=[("foo", "bar 的")],
+        client_session=ClientSession(
+            user="test",
+            extra_credential=[("foo", "bar 的")],
+        ),
     )
 
     req.post("SELECT 1")
@@ -737,7 +784,9 @@ def test_authentication_fail_retry(monkeypatch):
     req = TrinoRequest(
         host="coordinator",
         port=8080,
-        user="test",
+        client_session=ClientSession(
+            user="test",
+        ),
         http_scheme=constants.HTTPS,
         auth=kerberos_auth,
         max_attempts=attempts,
@@ -768,7 +817,12 @@ def test_5XX_error_retry(status_code, attempts, monkeypatch):
     monkeypatch.setattr(TrinoRequest.http.Session, "get", get_retry)
 
     req = TrinoRequest(
-        host="coordinator", port=8080, user="test", max_attempts=attempts
+        host="coordinator",
+        port=8080,
+        client_session=ClientSession(
+            user="test",
+        ),
+        max_attempts=attempts
     )
 
     req.post("URL")
@@ -792,7 +846,12 @@ def test_error_no_retry(status_code, monkeypatch):
     monkeypatch.setattr(TrinoRequest.http.Session, "get", get_retry)
 
     req = TrinoRequest(
-        host="coordinator", port=8080, user="test", max_attempts=3
+        host="coordinator",
+        port=8080,
+        client_session=ClientSession(
+            user="test",
+        ),
+        max_attempts=3,
     )
 
     req.post("URL")
@@ -859,12 +918,14 @@ def test_trino_query_response_headers(sample_get_response_data):
     req = TrinoRequest(
         host="coordinator",
         port=8080,
-        user="test",
-        source="test",
-        catalog="test",
-        schema="test",
+        client_session=ClientSession(
+            user="test",
+            source="test",
+            catalog="test",
+            schema="test",
+            properties={},
+        ),
         http_scheme="http",
-        session_properties={},
     )
 
     sql = 'execute my_stament using 1, 2, 3'
diff --git a/tests/unit/test_dbapi.py b/tests/unit/test_dbapi.py
index f847007..7b1c72c 100644
--- a/tests/unit/test_dbapi.py
+++ b/tests/unit/test_dbapi.py
@@ -232,5 +232,5 @@ def test_tags_are_set_when_specified(mock_client):
         conn.cursor().execute("SOME FAKE QUERY")
 
     # THEN
-    _, passed_client_tags = mock_client.TrinoRequest.call_args
+    _, passed_client_tags = mock_client.ClientSession.call_args
     assert passed_client_tags["client_tags"] == client_tags