Add a GHA workflow to run tests and pre-commit checks

.github/workflows/ci.yml

@@ -0,0 +1,91 @@
+name: test
+
+on:
+  pull_request:
+    types:
+      - opened
+      - reopened
+      - synchronize
+  push:
+    branches:
+      - main
+
+jobs:
+  changed-files:
+    runs-on: ubuntu-latest
+    outputs:
+      should_run_pytest: ${{ steps.check-pytest.outputs.src }}
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - id: check-pytest
+        name: Should run pytest
+        uses: dorny/paths-filter@v3
+        with:
+          filters: |
+            src:
+              - "async_batcher/**"
+              - "tests/**"
+              - "poetry.lock"
+
+  pytest:
+    needs: changed-files
+    if: ${{ needs.changed-files.outputs.should_run_pytest == 'true' }}
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ "ubuntu-latest", "macos-latest" ]
+        python-version: [ "3.8", "3.9", "3.10", "3.11", "3.12" ]
+    runs-on: ${{ matrix.os }}
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v4
+
+      - name: Set up python ${{ matrix.python-version }}
+        id: setup-python
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install Poetry
+        uses: snok/install-poetry@v1
+        with:
+          virtualenvs-create: true
+          virtualenvs-in-project: true
+
+      - name: Load cached venv
+        id: cached-poetry-dependencies
+        uses: actions/cache@v4
+        with:
+          path: .venv
+          key: venv-${{ runner.os }}-${{ steps.setup-python.outputs.python-version }}-${{ hashFiles('**/poetry.lock') }}
+
+      - name: Install dependencies
+        if: steps.cached-poetry-dependencies.outputs.cache-hit != 'true'
+        run: poetry install --no-interaction --no-root --all-extras
+
+      - name: Install library
+        run: poetry install --no-interaction --all-extras
+
+      - name: Run tests
+        run: |
+          source .venv/bin/activate
+          pytest tests/
+
+  linter:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v4
+
+      - name: Set up python 3.11
+        uses: actions/setup-python@v5
+        with:
+          python-version: 3.11
+
+      - name: Install pre-commit
+        run: pip install pre-commit
+
+      - name: Run pre-commit
+        run: pre-commit run --all-files

async_batcher/batcher.py

@@ -4,7 +4,7 @@
 import asyncio
 import logging
 from collections import namedtuple
-from typing import Generic, TypeVar, Union
+from typing import Generic, TypeVar
 
 T = TypeVar("T")
 S = TypeVar("S")
@@ -38,7 +38,7 @@ def __init__(
         self.max_batch_size = max_batch_size
         self.max_queue_time = max_queue_time
         self._queue = asyncio.Queue()
-        self._current_task: Union[asyncio.Task, None] = None
+        self._current_task: asyncio.Task | None = None
         self._should_stop = False
         self._force_stop = False
         self._is_running = False
@@ -102,7 +102,9 @@ async def batch_run(self):
                 if asyncio.iscoroutinefunction(self.process_batch):
                     results = await self.process_batch(batch=batch_items)
                 else:
-                    results = await asyncio.get_event_loop().run_in_executor(None, self.process_batch, batch_items)
+                    results = await asyncio.get_event_loop().run_in_executor(
+                        None, self.process_batch, batch_items
+                    )
                 if results is None:
                     results = [None] * len(batch)
                 if len(results) != len(batch):
@@ -115,13 +117,9 @@ async def batch_run(self):
                 for q_item, result in zip(batch, results):
                     q_item.future.set_result(result)
             elapsed_time = asyncio.get_event_loop().time() - started_at
-            self.logger.debug(
-                f"Processed batch of {len(batch)} elements"
-                f" in {elapsed_time} seconds."
-            )
+            self.logger.debug(f"Processed batch of {len(batch)} elements" f" in {elapsed_time} seconds.")
         self._is_running = False
 
-
     def stop(self, force: bool = False):
         """Stop the batcher asyncio task.
 
@@ -132,5 +130,9 @@ def stop(self, force: bool = False):
         if force:
             self._force_stop = True
         self._should_stop = True
-        if self._current_task and not self._current_task.done() and not self._current_task.get_loop().is_closed():
+        if (
+            self._current_task
+            and not self._current_task.done()
+            and not self._current_task.get_loop().is_closed()
+        ):
             self._current_task.get_loop().run_until_complete(self._current_task)

examples/dynamodb/main.py

@@ -5,9 +5,8 @@
 import aioboto3
 from async_batcher.aws.dynamodb.get import AsyncDynamoDbGetBatcher, GetItem
 from async_batcher.aws.dynamodb.write import AsyncDynamoDbWriteBatcher, WriteOperation
-from pydantic import BaseModel
-
 from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
 
 app = FastAPI()
 

examples/keras/main.py

@@ -1,13 +1,15 @@
 from __future__ import annotations
 
-import asyncio
 import gc
+from typing import TYPE_CHECKING
 
-import keras.src.engine.sequential
 import tensorflow as tf
 from async_batcher.batcher import AsyncBatcher
 from fastapi import FastAPI
 
+if TYPE_CHECKING:
+    import keras.src.engine.sequential
+
 
 class MlBatcher(AsyncBatcher[list[float], list[float]]):
     def __init__(self, model, *args, **kwargs):
@@ -24,10 +26,12 @@ def process_batch(self, batch: list[list[float]]) -> list[float]:
 
 batcher = MlBatcher(model=model, max_queue_time=0.001)
 
+
 @app.on_event("startup")
 async def startup_event():
     gc.freeze()
 
+
 @app.on_event("shutdown")
 def shutdown_event():
     batcher.stop()