odbc core initial work

Author: shaypal5

.github/workflows/ci-test.yml

@@ -22,18 +22,23 @@ jobs:
       matrix:
         python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
         os: ["ubuntu-latest", "macOS-latest", "windows-latest"]
-        backend: ["local", "db"]
+        backend: ["local", "mongodb", "mysql"]
         exclude:
           # ToDo: take if back when the connection become stable
           #  or resolve using `InMemoryMongoClient`
-          - { os: "macOS-latest", backend: "db" }
+          - { os: "macOS-latest", backend: "mongodb" }
+          - { os: "macOS-latest", backend: "mysql" }
     env:
-      CACHIER_TEST_HOST: "localhost"
-      CACHIER_TEST_PORT: "27017"
       #CACHIER_TEST_DB: "dummy_db"
       #CACHIER_TEST_USERNAME: "myuser"
       #CACHIER_TEST_PASSWORD: "yourpassword"
+      # CACHIER_MONGODB_TEST_HOST: "localhost"
+      CACHIER_MONGODB_TEST_PORT: "27017"
       CACHIER_TEST_VS_DOCKERIZED_MONGO: "true"
+      # CACHIER_MYSQL_TEST_PORT: "3306"
+      CACHIER_TEST_VS_DOCKERIZED_MYSQL: "true"
+      CACHIER_TEST_PYODBC_CONNECTION_STRING: "DRIVER={MySQL ODBC Driver};SERVER=localhost;PORT=3306;DATABASE=test;USER=root;PASSWORD=password;"
+
 
     steps:
       - uses: actions/checkout@v4
@@ -50,35 +55,45 @@ jobs:
 
       - name: Unit tests (local)
         if: matrix.backend == 'local'
-        run: pytest -m "not mongo"
+        run: pytest -m "not mongo and not mysql"
 
       - name: Setup docker (missing on MacOS)
-        if: runner.os == 'macOS' && matrix.backend == 'db'
+        if: runner.os == 'macOS' && (matrix.backend == 'mongodb' || matrix.backend == 'mysql')
         run: |
           brew install docker
           colima start
           # For testcontainers to find the Colima socket
           sudo ln -sf $HOME/.colima/default/docker.sock /var/run/docker.sock
       # ToDo: find a way to cache docker images
       #- name: Cache Container Images
-      #  if: matrix.backend == 'db'
+      #  if: matrix.backend == 'mongodb'
       #  uses: borda/cache-container-images-action@b32a5e804cb39af3c3d134fc03ab76eac0bfcfa9
       #  with:
       #    prefix-key: "mongo-db"
       #    images: mongo:latest
       - name: Start MongoDB in docker
-        if: matrix.backend == 'db'
+        if: matrix.backend == 'mongodb'
         run: |
           # start MongoDB in a container
-          docker run -d -p ${{ env.CACHIER_TEST_PORT }}:27017 --name mongodb mongo:latest
+          docker run -d -p ${{ env.CACHIER_MONGODB_TEST_PORT }}:27017 --name mongodb mongo:latest
           # wait for MongoDB to start, which is in average 5 seconds
           sleep 5
           # show running containers
           docker ps -a
-      - name: Unit tests (DB)
-        if: matrix.backend == 'db'
+      - name: Unit tests (MongoDB)
+        if: matrix.backend == 'mongodb'
         run: pytest -m "mongo"
 
+      - name: Start MySQL in docker
+        if: matrix.backend == 'mysql'
+        run: |
+          docker run -d -p ${{ env.CACHIER_MYSQL_TEST_PORT }}:3306 --name mysql -e MYSQL_ROOT_PASSWORD=password -e MYSQL_DATABASE=test mysql:latest
+          sleep 10
+          docker ps -a
+      - name: Unit tests (MySQL)
+        if: matrix.backend == 'mysql'
+        run: pytest -m "pyodbc"
+
       - name: "Upload coverage to Codecov"
         continue-on-error: true
         uses: codecov/codecov-action@v4

src/cachier/_types.py

@@ -6,4 +6,4 @@
 
 HashFunc = Callable[..., str]
 Mongetter = Callable[[], "pymongo.collection.Collection"]
-Backend = Literal["pickle", "mongo", "memory"]
+Backend = Literal["pickle", "mongo", "odbc", "memory"]

src/cachier/core.py

@@ -27,6 +27,7 @@
 from .cores.base import RecalculationNeeded, _BaseCore
 from .cores.memory import _MemoryCore
 from .cores.mongo import _MongoCore
+from .cores.odbc import _OdbcCore
 from .cores.pickle import _PickleCore
 
 MAX_WORKERS_ENVAR_NAME = "CACHIER_MAX_WORKERS"
@@ -110,6 +111,8 @@ def cachier(
     hash_params: Optional[HashFunc] = None,
     backend: Optional[Backend] = None,
     mongetter: Optional[Mongetter] = None,
+    odbc_connection_string: Optional[str] = None,
+    odbc_table_name: Optional[str] = None,
     stale_after: Optional[datetime.timedelta] = None,
     next_time: Optional[bool] = None,
     cache_dir: Optional[Union[str, os.PathLike]] = None,
@@ -137,13 +140,21 @@ def cachier(
     hash_params : callable, optional
     backend : str, optional
         The name of the backend to use. Valid options currently include
-        'pickle', 'mongo' and 'memory'. If not provided, defaults to
+        'pickle', 'mongo', 'odbc' and 'memory'. If not provided, defaults to
         'pickle' unless the 'mongetter' argument is passed, in which
-        case the mongo backend is automatically selected.
+        case the mongo backend is automatically selected, or the
+        'odbc_connection_string' argument is passed, in which case the odbc
+        backend is automatically selected.
     mongetter : callable, optional
         A callable that takes no arguments and returns a pymongo.Collection
         object with writing permissions. If unset a local pickle cache is used
         instead.
+    odbc_connection_string : str, optional
+        A connection string to an ODBC database. If provided, the ODBC core
+        will be used.
+    odbc_table_name : str, optional
+        The name of the table to use in the ODBC database. If not provided,
+        defaults to 'cachier'.
     stale_after : datetime.timedelta, optional
         The time delta after which a cached result is considered stale. Calls
         made after the result goes stale will trigger a recalculation of the
@@ -190,6 +201,8 @@ def cachier(
     # Override the backend parameter if a mongetter is provided.
     if callable(mongetter):
         backend = "mongo"
+    if odbc_connection_string is not None:
+        backend = "odbc"
     core: _BaseCore
     if backend == "pickle":
         core = _PickleCore(
@@ -205,6 +218,13 @@ def cachier(
             mongetter=mongetter,
             wait_for_calc_timeout=wait_for_calc_timeout,
         )
+    elif backend == "odbc":
+        core = _OdbcCore(
+            hash_func=hash_func,
+            wait_for_calc_timeout=wait_for_calc_timeout,
+            connection_string=odbc_connection_string,
+            table_name=odbc_table_name,
+        )
     elif backend == "memory":
         core = _MemoryCore(
             hash_func=hash_func, wait_for_calc_timeout=wait_for_calc_timeout

src/cachier/cores/odbc.py

@@ -0,0 +1,117 @@
+"""A pyodbc-based caching core for cachier."""
+
+# This file is part of Cachier.
+# https://github.com/python-cachier/cachier
+
+# Licensed under the MIT license:
+# http://www.opensource.org/licenses/MIT-license
+# Copyright (c) 2016, Shay Palachy <[email protected]>
+
+# standard library imports
+import pickle
+import time
+import datetime
+
+pyodbc = None
+# third party imports
+with suppress(ImportError):
+    import pyodbc
+
+# local imports
+from .base import _BaseCore, RecalculationNeeded
+
+class _OdbcCore(_BaseCore):
+
+    def __init__(
+            self,
+            hash_func,
+            wait_for_calc_timeout,
+            connection_string,
+            table_name,
+    ):
+        if "pyodbc" not in sys.modules:
+            warnings.warn(
+                "`pyodbc` was not found. pyodbc cores will not function.",
+                ImportWarning,
+                stacklevel=2,
+            )  # pragma: no cover
+        super().__init__(hash_func, wait_for_calc_timeout)
+        self.connection_string = connection_string
+        self.table_name = table_name
+        self.ensure_table_exists()
+
+    def ensure_table_exists(self):
+        with pyodbc.connect(self.connection_string) as conn:
+            cursor = conn.cursor()
+            cursor.execute(f"""
+                IF NOT EXISTS (SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = N'{self.table_name}')
+                BEGIN
+                    CREATE TABLE {self.table_name} (
+                        key NVARCHAR(255),
+                        value VARBINARY(MAX),
+                        time DATETIME,
+                        being_calculated BIT,
+                        PRIMARY KEY (key)
+                    );
+                END
+            """)
+            conn.commit()
+
+    def get_entry_by_key(self, key):
+        with pyodbc.connect(self.connection_string) as conn:
+            cursor = conn.cursor()
+            cursor.execute(f"SELECT value, time, being_calculated FROM {self.table_name} WHERE key = ?", key)
+            row = cursor.fetchone()
+            if row:
+                return {
+                    "value": pickle.loads(row.value),
+                    "time": row.time,
+                    "being_calculated": row.being_calculated,
+                }
+            return None
+
+    def set_entry(self, key, func_res):
+        with pyodbc.connect(self.connection_string) as conn:
+            cursor = conn.cursor()
+            cursor.execute(f"""
+                MERGE INTO {self.table_name} USING (SELECT 1 AS dummy) AS src ON (key = ?)
+                WHEN MATCHED THEN
+                    UPDATE SET value = ?, time = GETDATE(), being_calculated = 0
+                WHEN NOT MATCHED THEN
+                    INSERT (key, value, time, being_calculated) VALUES (?, ?, GETDATE(), 0);
+            """, key, pickle.dumps(func_res), key, pickle.dumps(func_res))
+            conn.commit()
+
+    def mark_entry_being_calculated(self, key):
+        with pyodbc.connect(self.connection_string) as conn:
+            cursor = conn.cursor()
+            cursor.execute(f"UPDATE {self.table_name} SET being_calculated = 1 WHERE key = ?", key)
+            conn.commit()
+
+    def mark_entry_not_calculated(self, key):
+        with pyodbc.connect(self.connection_string) as conn:
+            cursor = conn.cursor()
+            cursor.execute(f"UPDATE {self.table_name} SET being_calculated = 0 WHERE key = ?", key)
+            conn.commit()
+
+    def wait_on_entry_calc(self, key):
+        start_time = datetime.datetime.now()
+        while True:
+            entry = self.get_entry_by_key(key)
+            if entry and not entry['being_calculated']:
+                return entry['value']
+            if (datetime.datetime.now() - start_time).total_seconds() > self.wait_for_calc_timeout:
+                raise RecalculationNeeded()
+            time.sleep(1)
+
+    def clear_cache(self):
+        with pyodbc.connect(self.connection_string) as conn:
+            cursor = conn.cursor()
+            cursor.execute(f"DELETE FROM {self.table_name}")
+            conn.commit()
+
+    def clear_being_calculated(self):
+        with pyodbc.connect(self.connection_string) as conn:
+            cursor = conn.cursor()
+            cursor.execute(f"UPDATE {self.table_name} SET being_calculated = 0")
+            conn.commit()

tests/test_mongo_core.py

@@ -26,8 +26,8 @@
 
 
 class CfgKey:
-    HOST = "TEST_HOST"
-    PORT = "TEST_PORT"
+    HOST = "MONGODB_TEST_HOST"
+    PORT = "MONGODB_TEST_PORT"
     # UNAME = "TEST_USERNAME"
     # PWD = "TEST_PASSWORD"
     # DB = "TEST_DB"

tests/test_odbc_core.py

@@ -0,0 +1,85 @@
+"""Testing the MongoDB core of cachier."""
+
+# standard library imports
+import datetime
+from time import sleep
+
+# third party imports
+import pytest
+from birch import Birch  # type: ignore[import-not-found]
+
+# local imports
+from cachier import cachier
+# from cachier.cores.base import RecalculationNeeded
+# from cachier.cores.odbc import _OdbcCore
+
+
+class CfgKey:
+    """Configuration keys for testing."""
+    TEST_VS_DOCKERIZED_MYSQL = "TEST_VS_DOCKERIZED_MYSQL"
+    TEST_PYODBC_CONNECTION_STRING = "TEST_PYODBC_CONNECTION_STRING"
+
+
+CFG = Birch(
+    namespace="cachier",
+    defaults={CfgKey.TEST_VS_DOCKERIZED_MYSQL: False},
+)
+
+# Configuration for ODBC connection for tests
+CONCT_STR = CFG.mget(CfgKey.TEST_PYODBC_CONNECTION_STRING)
+# TABLE_NAME = "test_cache_table"
+
+
+@pytest.mark.odbc
+def test_odbc_entry_creation_and_retrieval(odbc_core):
+    """Test inserting and retrieving an entry from ODBC cache."""
+
+    @cachier(backend='odbc', odbc_connection_string=CONCT_STR)
+    def sample_function(arg_1, arg_2):
+        return arg_1 + arg_2
+
+    sample_function.clear_cache()
+    assert sample_function(1, 2) == 3  # Test cache miss and insertion
+    assert sample_function(1, 2) == 3  # Test cache hit
+
+
+@pytest.mark.odbc
+def test_odbc_stale_after(odbc_core):
+    """Test ODBC core handling stale_after parameter."""
+    stale_after = datetime.timedelta(seconds=1)
+
+    @cachier(backend='odbc', odbc_connection_string=CONCT_STR, stale_after=stale_after)
+    def stale_test_function(arg_1, arg_2):
+        return arg_1 + arg_2 + datetime.datetime.now().timestamp()  # Add timestamp to ensure unique values
+
+    initial_value = stale_test_function(5, 10)
+    sleep(2)  # Wait for the entry to become stale
+    assert stale_test_function(5, 10) != initial_value  # Should recompute since stale
+
+
+@pytest.mark.odbc
+def test_odbc_clear_cache(odbc_core):
+    """Test clearing the ODBC cache."""
+    @cachier(backend='odbc', odbc_connection_string=CONCT_STR)
+    def clearable_function(arg):
+        return arg
+
+    clearable_function.clear_cache()  # Ensure clean state
+    assert clearable_function(3) == 3  # Populate cache
+    clearable_function.clear_cache()  # Clear cache
+    # The next call should recompute result indicating that cache was cleared
+    assert clearable_function(3) == 3
+
+
+@pytest.mark.odbc
+def test_odbc_being_calculated_flag(odbc_core):
+    """Test handling of 'being_calculated' flag in ODBC core."""
+    @cachier(backend='odbc', odbc_connection_string=CONCT_STR)
+    def slow_function(arg):
+        sleep(2)  # Simulate long computation
+        return arg * 2
+
+    slow_function.clear_cache()
+    result1 = slow_function(4)
+    result2 = slow_function(4)  # Should hit cache, not wait for recalculation
+    assert result1 == result2