FIX: Fix for windows workflows

Author: nmacholl

.github/workflows/release.yml

@@ -26,7 +26,7 @@ jobs:
 
       - name: Install poetry
         run: |
-          python -m pip install --upgrade pip poetry==1.3.2
+          python -m pip install --upgrade pip poetry==1.5.1
 
       - name: Prepare release
         id: vars
@@ -62,7 +62,7 @@ jobs:
 
       - name: Install poetry
         run: |
-          python -m pip install --upgrade pip poetry==1.3.2
+          python -m pip install --upgrade pip poetry==1.5.1
 
       - name: Build
         run: |

.github/workflows/test.yml

@@ -23,22 +23,21 @@ jobs:
           python-version: ${{ matrix.python-version }}
 
       - name: Install poetry
-        run: python -m pip install --upgrade pip poetry==1.3.2
-
-      - name: Load cached venv
-        id: cached-poetry-dependencies
-        uses: actions/cache@v3
-        with:
-          path: .venv
-          key: venv-${{ runner.os }}-${{ steps.setup-python.outputs.python-version }}-${{ hashFiles('**/poetry.lock') }}
+        run: python -m pip install --upgrade pip poetry==1.5.1
 
       - name: Install dependencies
         if: steps.cached-poetry-dependencies.outputs.cache-hit != 'true'
         run: scripts/build.sh
+        shell: bash
 
       - name: Run tests in release mode
+        timeout-minutes: 5
         if: ${{ github.ref == 'refs/heads/main' }}
         run: scripts/test.sh --release
+        shell: bash
+
       - name: Run tests
+        timeout-minutes: 5
         if: ${{ github.ref != 'refs/heads/main' }}
         run: scripts/test.sh
+        shell: bash

databento/live/client.py

@@ -128,10 +128,10 @@ def __next__(self) -> DBNRecord:
         if self._dbn_queue is None:
             raise ValueError("iteration has not started")
 
-        while not self._session.is_disconnected() or not self._dbn_queue.empty():
+        while not self._session.is_disconnected() or self._dbn_queue.qsize() > 0:
             try:
-                record = self._dbn_queue.get(timeout=0.001)
-            except (futures.TimeoutError, queue.Empty):
+                record = self._dbn_queue.get(block=False)
+            except queue.Empty:
                 continue
             else:
                 logger.debug(
@@ -476,9 +476,6 @@ def block_for_close(
         wait_for_close
 
         """
-        if not self.is_connected():
-            return
-
         try:
             asyncio.run_coroutine_threadsafe(
                 self._shutdown(),
@@ -520,9 +517,6 @@ async def wait_for_close(
         block_for_close
 
         """
-        if not self.is_connected():
-            return
-
         waiter = asyncio.wrap_future(
             asyncio.run_coroutine_threadsafe(
                 self._shutdown(),
@@ -549,5 +543,4 @@ async def _shutdown(self) -> None:
         """
         if self._session is None:
             return
-        self._dataset = ""  # reset dataset for client reuse
         await self._session.wait_for_close()

databento/live/session.py

@@ -3,17 +3,21 @@
 import logging
 import queue
 import struct
-import threading
 from typing import IO, Callable, Iterable, List, Optional, Set, Union
 
 import databento_dbn
 
-from databento.common.enums import Dataset, Schema, SType
+from databento.common.enums import Dataset
+from databento.common.enums import Schema
+from databento.common.enums import SType
 from databento.common.error import BentoError
 from databento.common.symbology import ALL_SYMBOLS
-from databento.live import AUTH_TIMEOUT_SECONDS, CONNECT_TIMEOUT_SECONDS, DBNRecord
+from databento.live import AUTH_TIMEOUT_SECONDS
+from databento.live import CONNECT_TIMEOUT_SECONDS
+from databento.live import DBNRecord
 from databento.live.protocol import DatabentoLiveProtocol
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -258,7 +262,6 @@ def __init__(
         self._ts_out = ts_out
         self._protocol_factory = protocol_factory
 
-        self._lock = threading.Lock()
         self._transport: Optional[asyncio.Transport] = None
         self._protocol: Optional[_SessionProtocol] = None
 
@@ -276,12 +279,11 @@ def is_authenticated(self) -> bool:
         """
         if self._protocol is None:
             return False
-        with self._lock:
-            try:
-                self._protocol.authenticated.result()
-            except (asyncio.InvalidStateError, asyncio.CancelledError, BentoError):
-                return False
-            return True
+        try:
+            self._protocol.authenticated.result()
+        except (asyncio.InvalidStateError, asyncio.CancelledError, BentoError):
+            return False
+        return True
 
     def is_disconnected(self) -> bool:
         """
@@ -294,12 +296,7 @@ def is_disconnected(self) -> bool:
         """
         if self._protocol is None:
             return True
-        with self._lock:
-            try:
-                self._protocol.disconnected.result()
-            except (asyncio.InvalidStateError, asyncio.CancelledError, Exception):
-                return False
-            return True
+        return self._protocol.disconnected.done()
 
     def is_reading(self) -> bool:
         """
@@ -312,8 +309,7 @@ def is_reading(self) -> bool:
         """
         if self._transport is None:
             return False
-        with self._lock:
-            return self._transport.is_reading()
+        return self._transport.is_reading()
 
     def is_started(self) -> bool:
         """
@@ -326,8 +322,7 @@ def is_started(self) -> bool:
         """
         if self._protocol is None:
             return False
-        with self._lock:
-            return self._protocol.started.is_set()
+        return self._protocol.started.is_set()
 
     @property
     def metadata(self) -> Optional[databento_dbn.Metadata]:
@@ -356,9 +351,8 @@ def abort(self) -> None:
         """
         if self._transport is None:
             return
-        with self._lock:
-            self._transport.abort()
-            self._protocol = None
+        self._transport.abort()
+        self._protocol = None
 
     def close(self) -> None:
         """
@@ -367,10 +361,9 @@ def close(self) -> None:
         """
         if self._transport is None:
             return
-        with self._lock:
-            if self._transport.can_write_eof():
-                self._transport.write_eof()
-            self._transport.close()
+        if self._transport.can_write_eof():
+            self._loop.call_soon_threadsafe(self._transport.write_eof)
+        self._loop.call_soon_threadsafe(self._transport.close)
 
     def subscribe(
         self,
@@ -421,8 +414,7 @@ def resume_reading(self) -> None:
         """
         if self._transport is None:
             return
-        with self._lock:
-            self._transport.resume_reading()
+        self._loop.call_soon_threadsafe(self._transport.resume_reading)
 
     def start(self) -> None:
         """
@@ -445,10 +437,16 @@ async def wait_for_close(self) -> None:
         """
         if self._protocol is None:
             return
+
         await self._protocol.disconnected
+        disconnect_exc = self._protocol.disconnected.exception()
+
         await self._protocol.wait_for_processing()
         self._protocol = self._transport = None
 
+        if disconnect_exc is not None:
+            raise BentoError(disconnect_exc)
+
     def _connect(
         self,
         dataset: Union[Dataset, str],
@@ -521,6 +519,5 @@ async def _connect_task(
             "authentication with remote gateway completed",
         )
 
-        with self._lock:
-            self._transport = transport
-            self._protocol = protocol
+        self._transport = transport
+        self._protocol = protocol

pyproject.toml

@@ -33,10 +33,12 @@ requests = ">=2.24.0"
 zstandard = ">=0.21.0"
 
 [tool.poetry.group.dev.dependencies]
-mypy = "1.2.0"
+black = "^23.3.0"
+mypy = "1.3.0"
 pytest = ">=7.3.1"
 pytest-asyncio = ">=0.21.0"
 pytest-mock = ">=3.10.0"
+ruff = "^0.0.270"
 types-requests = "^2.30.0.0"
 
 [build-system]

scripts/build.sh

@@ -1,2 +1,2 @@
-#! /usr/bin/env bash
+#!/usr/bin/env bash
 poetry install --with=dev --no-interaction --no-root

scripts/lint.sh

@@ -1,4 +1,4 @@
-#! /usr/bin/env bash
+#!/usr/bin/env bash
 echo $(mypy --version)
 echo Running mypy...
 poetry run mypy databento examples tests

scripts/test.sh

@@ -1,2 +1,2 @@
-#! /usr/bin/env bash
+#!/usr/bin/env bash
 poetry run pytest tests . $@

tests/conftest.py

@@ -5,6 +5,7 @@
 import string
 from typing import AsyncGenerator, Callable, Generator, Iterable
 
+import databento.live
 import pytest
 import pytest_asyncio
 from databento.common.enums import Schema
@@ -154,7 +155,6 @@ def fixture_test_api_key() -> str:
 
 @pytest_asyncio.fixture(name="mock_live_server")
 async def fixture_mock_live_server(
-    event_loop: asyncio.AbstractEventLoop,
     test_api_key: str,
     caplog: pytest.LogCaptureFixture,
     unused_tcp_port: int,
@@ -168,11 +168,20 @@ async def fixture_mock_live_server(
     MockLiveServer
 
     """
-    event_loop.set_debug(True)
     monkeypatch.setenv(
         name="DATABENTO_API_KEY",
         value=test_api_key,
     )
+    monkeypatch.setattr(
+        databento.live,
+        "AUTH_TIMEOUT_SECONDS",
+        1,
+    )
+    monkeypatch.setattr(
+        databento.live,
+        "CONNECT_TIMEOUT_SECONDS",
+        1,
+    )
 
     with caplog.at_level("DEBUG"):
         mock_live_server = await MockLiveServer.create(