SQLite Support (#441)

This PR adds support for using SQLite as the system database. This is
intended to be used for lightweight development and testing, allowing
users to get started with DBOS or test their DBOS apps without needing a
Postgres database. It may also be useful for embedded applications.

The SQLite system database is 100% feature complete. To minimize code
duplication, we use SQLAlchemy to implement virtually all system
database operations cross-dialect. The only operations that need custom
implementations in Postgres and SQLite are:

1. Database creation/migration/teardown.
2. Notification management (using LISTEN/NOTIFY in Postgres and polling
in SQLite).
3. Exception categorization (differentiating primary key conflicts from
other errors, for example).

The test matrix has been reconfigured to run all tests against both
Postgres and SQLite. Only a handful of Postgres-specific tests run on
Postgres only.

DBOS now uses SQLite by default if no database URL is provided to it.
This is to make it easier to get started on DBOS—you can now try
it from anywhere without Postgres.

Author: kraftp

.github/workflows/unit-test.yml

@@ -23,6 +23,7 @@ jobs:
       fail-fast: false
       matrix:
         python-version: ['3.9', '3.10', '3.11', '3.12', '3.13']
+        dbos-database: ['SQLITE', 'POSTGRES']
     services:
       # Postgres service container
       postgres:
@@ -98,4 +99,5 @@ jobs:
       working-directory: ./
       env:
         PGPASSWORD: a!b@c$d()e*_,/:;=?@ff[]22
-        GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        DBOS_DATABASE: ${{ matrix.dbos-database }}

.gitignore

@@ -169,3 +169,6 @@ cython_debug/
 # IDE files
 .vscode/
 .idea/
+
+# Sqlite
+*.sqlite

dbos/_app_db.py

@@ -1,11 +1,14 @@
+from abc import ABC, abstractmethod
 from typing import Any, Dict, List, Optional, TypedDict
 
+import psycopg
 import sqlalchemy as sa
-import sqlalchemy.dialects.postgresql as pg
 from sqlalchemy import inspect, text
 from sqlalchemy.exc import DBAPIError
 from sqlalchemy.orm import Session, sessionmaker
 
+from dbos._migration import get_sqlite_timestamp_expr
+
 from . import _serialization
 from ._error import DBOSUnexpectedStepError, DBOSWorkflowConflictIDError
 from ._logger import dbos_logger
@@ -29,7 +32,7 @@ class RecordedResult(TypedDict):
     error: Optional[str]  # JSON (jsonpickle)
 
 
-class ApplicationDatabase:
+class ApplicationDatabase(ABC):
 
     def __init__(
         self,
@@ -38,95 +41,37 @@ def __init__(
         engine_kwargs: Dict[str, Any],
         debug_mode: bool = False,
     ):
-        app_db_url = sa.make_url(database_url).set(drivername="postgresql+psycopg")
-
-        if engine_kwargs is None:
-            engine_kwargs = {}
-
-        self.engine = sa.create_engine(
-            app_db_url,
-            **engine_kwargs,
-        )
+        self.engine = self._create_engine(database_url, engine_kwargs)
         self._engine_kwargs = engine_kwargs
         self.sessionmaker = sessionmaker(bind=self.engine)
         self.debug_mode = debug_mode
 
-    def run_migrations(self) -> None:
-        if self.debug_mode:
-            dbos_logger.warning(
-                "Application database migrations are skipped in debug mode."
-            )
-            return
-        # Check if the database exists
-        app_db_url = self.engine.url
-        postgres_db_engine = sa.create_engine(
-            app_db_url.set(database="postgres"),
-            **self._engine_kwargs,
-        )
-        with postgres_db_engine.connect() as conn:
-            conn.execution_options(isolation_level="AUTOCOMMIT")
-            if not conn.execute(
-                sa.text("SELECT 1 FROM pg_database WHERE datname=:db_name"),
-                parameters={"db_name": app_db_url.database},
-            ).scalar():
-                conn.execute(sa.text(f"CREATE DATABASE {app_db_url.database}"))
-        postgres_db_engine.dispose()
-
-        # Create the dbos schema and transaction_outputs table in the application database
-        with self.engine.begin() as conn:
-            # Check if schema exists first
-            schema_exists = conn.execute(
-                sa.text(
-                    "SELECT 1 FROM information_schema.schemata WHERE schema_name = :schema_name"
-                ),
-                parameters={"schema_name": ApplicationSchema.schema},
-            ).scalar()
-
-            if not schema_exists:
-                schema_creation_query = sa.text(
-                    f"CREATE SCHEMA {ApplicationSchema.schema}"
-                )
-                conn.execute(schema_creation_query)
-
-        inspector = inspect(self.engine)
-        if not inspector.has_table(
-            "transaction_outputs", schema=ApplicationSchema.schema
-        ):
-            ApplicationSchema.metadata_obj.create_all(self.engine)
-        else:
-            columns = inspector.get_columns(
-                "transaction_outputs", schema=ApplicationSchema.schema
-            )
-            column_names = [col["name"] for col in columns]
+    @abstractmethod
+    def _create_engine(
+        self, database_url: str, engine_kwargs: Dict[str, Any]
+    ) -> sa.Engine:
+        """Create a database engine specific to the database type."""
+        pass
 
-            if "function_name" not in column_names:
-                # Column missing, alter table to add it
-                with self.engine.connect() as conn:
-                    conn.execute(
-                        text(
-                            f"""
-                        ALTER TABLE {ApplicationSchema.schema}.transaction_outputs
-                        ADD COLUMN function_name TEXT NOT NULL DEFAULT '';
-                        """
-                        )
-                    )
-                    conn.commit()
+    @abstractmethod
+    def run_migrations(self) -> None:
+        """Run database migrations specific to the database type."""
+        pass
 
     def destroy(self) -> None:
         self.engine.dispose()
 
-    @staticmethod
     def record_transaction_output(
-        session: Session, output: TransactionResultInternal
+        self, session: Session, output: TransactionResultInternal
     ) -> None:
         try:
             session.execute(
-                pg.insert(ApplicationSchema.transaction_outputs).values(
+                sa.insert(ApplicationSchema.transaction_outputs).values(
                     workflow_uuid=output["workflow_uuid"],
                     function_id=output["function_id"],
                     output=output["output"],
                     error=None,
-                    txn_id=sa.text("(select pg_current_xact_id_if_assigned()::text)"),
+                    txn_id="",
                     txn_snapshot=output["txn_snapshot"],
                     executor_id=(
                         output["executor_id"] if output["executor_id"] else None
@@ -135,7 +80,7 @@ def record_transaction_output(
                 )
             )
         except DBAPIError as dbapi_error:
-            if dbapi_error.orig.sqlstate == "23505":  # type: ignore
+            if self._is_unique_constraint_violation(dbapi_error):
                 raise DBOSWorkflowConflictIDError(output["workflow_uuid"])
             raise
 
@@ -145,14 +90,12 @@ def record_transaction_error(self, output: TransactionResultInternal) -> None:
         try:
             with self.engine.begin() as conn:
                 conn.execute(
-                    pg.insert(ApplicationSchema.transaction_outputs).values(
+                    sa.insert(ApplicationSchema.transaction_outputs).values(
                         workflow_uuid=output["workflow_uuid"],
                         function_id=output["function_id"],
                         output=None,
                         error=output["error"],
-                        txn_id=sa.text(
-                            "(select pg_current_xact_id_if_assigned()::text)"
-                        ),
+                        txn_id="",
                         txn_snapshot=output["txn_snapshot"],
                         executor_id=(
                             output["executor_id"] if output["executor_id"] else None
@@ -161,7 +104,7 @@ def record_transaction_error(self, output: TransactionResultInternal) -> None:
                     )
                 )
         except DBAPIError as dbapi_error:
-            if dbapi_error.orig.sqlstate == "23505":  # type: ignore
+            if self._is_unique_constraint_violation(dbapi_error):
                 raise DBOSWorkflowConflictIDError(output["workflow_uuid"])
             raise
 
@@ -283,3 +226,197 @@ def garbage_collect(
                 )
 
             c.execute(delete_query)
+
+    @abstractmethod
+    def _is_unique_constraint_violation(self, dbapi_error: DBAPIError) -> bool:
+        """Check if the error is a unique constraint violation."""
+        pass
+
+    @abstractmethod
+    def _is_serialization_error(self, dbapi_error: DBAPIError) -> bool:
+        """Check if the error is a serialization/concurrency error."""
+        pass
+
+    @staticmethod
+    def create(
+        database_url: str,
+        engine_kwargs: Dict[str, Any],
+        debug_mode: bool = False,
+    ) -> "ApplicationDatabase":
+        """Factory method to create the appropriate ApplicationDatabase implementation based on URL."""
+        if database_url.startswith("sqlite"):
+            return SQLiteApplicationDatabase(
+                database_url=database_url,
+                engine_kwargs=engine_kwargs,
+                debug_mode=debug_mode,
+            )
+        else:
+            # Default to PostgreSQL for postgresql://, postgres://, or other URLs
+            return PostgresApplicationDatabase(
+                database_url=database_url,
+                engine_kwargs=engine_kwargs,
+                debug_mode=debug_mode,
+            )
+
+
+class PostgresApplicationDatabase(ApplicationDatabase):
+    """PostgreSQL-specific implementation of ApplicationDatabase."""
+
+    def _create_engine(
+        self, database_url: str, engine_kwargs: Dict[str, Any]
+    ) -> sa.Engine:
+        """Create a PostgreSQL engine."""
+        app_db_url = sa.make_url(database_url).set(drivername="postgresql+psycopg")
+
+        if engine_kwargs is None:
+            engine_kwargs = {}
+
+        # TODO: Make the schema dynamic so this isn't needed
+        ApplicationSchema.transaction_outputs.schema = "dbos"
+
+        return sa.create_engine(
+            app_db_url,
+            **engine_kwargs,
+        )
+
+    def run_migrations(self) -> None:
+        if self.debug_mode:
+            dbos_logger.warning(
+                "Application database migrations are skipped in debug mode."
+            )
+            return
+        # Check if the database exists
+        app_db_url = self.engine.url
+        postgres_db_engine = sa.create_engine(
+            app_db_url.set(database="postgres"),
+            **self._engine_kwargs,
+        )
+        with postgres_db_engine.connect() as conn:
+            conn.execution_options(isolation_level="AUTOCOMMIT")
+            if not conn.execute(
+                sa.text("SELECT 1 FROM pg_database WHERE datname=:db_name"),
+                parameters={"db_name": app_db_url.database},
+            ).scalar():
+                conn.execute(sa.text(f"CREATE DATABASE {app_db_url.database}"))
+        postgres_db_engine.dispose()
+
+        # Create the dbos schema and transaction_outputs table in the application database
+        with self.engine.begin() as conn:
+            # Check if schema exists first
+            schema_exists = conn.execute(
+                sa.text(
+                    "SELECT 1 FROM information_schema.schemata WHERE schema_name = :schema_name"
+                ),
+                parameters={"schema_name": ApplicationSchema.schema},
+            ).scalar()
+
+            if not schema_exists:
+                schema_creation_query = sa.text(
+                    f"CREATE SCHEMA {ApplicationSchema.schema}"
+                )
+                conn.execute(schema_creation_query)
+
+        inspector = inspect(self.engine)
+        if not inspector.has_table(
+            "transaction_outputs", schema=ApplicationSchema.schema
+        ):
+            ApplicationSchema.metadata_obj.create_all(self.engine)
+        else:
+            columns = inspector.get_columns(
+                "transaction_outputs", schema=ApplicationSchema.schema
+            )
+            column_names = [col["name"] for col in columns]
+
+            if "function_name" not in column_names:
+                # Column missing, alter table to add it
+                with self.engine.connect() as conn:
+                    conn.execute(
+                        text(
+                            f"""
+                        ALTER TABLE {ApplicationSchema.schema}.transaction_outputs
+                        ADD COLUMN function_name TEXT NOT NULL DEFAULT '';
+                        """
+                        )
+                    )
+                    conn.commit()
+
+    def _is_unique_constraint_violation(self, dbapi_error: DBAPIError) -> bool:
+        """Check if the error is a unique constraint violation in PostgreSQL."""
+        return dbapi_error.orig.sqlstate == "23505"  # type: ignore
+
+    def _is_serialization_error(self, dbapi_error: DBAPIError) -> bool:
+        """Check if the error is a serialization/concurrency error in PostgreSQL."""
+        # 40001: serialization_failure (MVCC conflict)
+        # 40P01: deadlock_detected
+        driver_error = dbapi_error.orig
+        return (
+            driver_error is not None
+            and isinstance(driver_error, psycopg.OperationalError)
+            and driver_error.sqlstate in ("40001", "40P01")
+        )
+
+
+class SQLiteApplicationDatabase(ApplicationDatabase):
+    """SQLite-specific implementation of ApplicationDatabase."""
+
+    def _create_engine(
+        self, database_url: str, engine_kwargs: Dict[str, Any]
+    ) -> sa.Engine:
+        """Create a SQLite engine."""
+        # TODO: Make the schema dynamic so this isn't needed
+        ApplicationSchema.transaction_outputs.schema = None
+        return sa.create_engine(database_url)
+
+    def run_migrations(self) -> None:
+        if self.debug_mode:
+            dbos_logger.warning(
+                "Application database migrations are skipped in debug mode."
+            )
+            return
+
+        with self.engine.begin() as conn:
+            # Check if table exists
+            result = conn.execute(
+                sa.text(
+                    "SELECT name FROM sqlite_master WHERE type='table' AND name='transaction_outputs'"
+                )
+            ).fetchone()
+
+            if result is None:
+                # Create the table with proper SQLite syntax
+                conn.execute(
+                    sa.text(
+                        f"""
+                        CREATE TABLE transaction_outputs (
+                            workflow_uuid TEXT NOT NULL,
+                            function_id INTEGER NOT NULL,
+                            output TEXT,
+                            error TEXT,
+                            txn_id TEXT,
+                            txn_snapshot TEXT NOT NULL,
+                            executor_id TEXT,
+                            function_name TEXT NOT NULL DEFAULT '',
+                            created_at BIGINT NOT NULL DEFAULT {get_sqlite_timestamp_expr()},
+                            PRIMARY KEY (workflow_uuid, function_id)
+                        )
+                        """
+                    )
+                )
+                # Create the index
+                conn.execute(
+                    sa.text(
+                        "CREATE INDEX transaction_outputs_created_at_index ON transaction_outputs (created_at)"
+                    )
+                )
+
+    def _is_unique_constraint_violation(self, dbapi_error: DBAPIError) -> bool:
+        """Check if the error is a unique constraint violation in SQLite."""
+        return "UNIQUE constraint failed" in str(dbapi_error.orig)
+
+    def _is_serialization_error(self, dbapi_error: DBAPIError) -> bool:
+        """Check if the error is a serialization/concurrency error in SQLite."""
+        # SQLite database is locked or busy errors
+        error_msg = str(dbapi_error.orig).lower()
+        return (
+            "database is locked" in error_msg or "database table is locked" in error_msg
+        )