Generic SQL (using pyodbc) core - initial work

.github/workflows/ci-test.yml

@@ -22,18 +22,22 @@ 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_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 +54,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

pyproject.toml

@@ -135,6 +135,7 @@ blank = true
 
 [tool.pytest.ini_options]
 testpaths = [
+  "src/cachier",
   "cachier",
   "tests",
 ]

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,141 @@
+"""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 datetime
+import pickle
+import sys
+import time
+import warnings
+from contextlib import suppress
+
+pyodbc = None
+# third party imports
+with suppress(ImportError):
+    import pyodbc
+
+# local imports
+from .base import RecalculationNeeded, _BaseCore
+
+
+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"