Improve checking of in-memory SQLite database.

Author: danielballan

tiled/_tests/test_connection_pool.py

@@ -0,0 +1,21 @@
+from typing import Union
+
+import pytest
+from sqlalchemy.engine import URL, make_url
+
+from tiled.server.connection_pool import is_memory_sqlite
+
+
+@pytest.mark.parametrize(
+    ("uri", "expected"),
+    [
+        ("sqlite://", True),  # accepts str
+        (make_url("sqlite://"), True),  # accepts URL
+        ("sqlite:///:memory:", True),
+        ("sqlite:///file::memory:?cache=shared", True),
+        ("sqlite:////tmp/example.db", False),
+    ],
+)
+def test_is_memory_sqlite(uri: Union[str, URL], expected: bool):
+    actual = is_memory_sqlite(uri)
+    assert actual is expected

tiled/catalog/adapter.py

@@ -68,7 +68,11 @@
     ZARR_MIMETYPE,
 )
 from ..query_registration import QueryTranslationRegistry
-from ..server.connection_pool import close_database_connection_pool, get_database_engine
+from ..server.connection_pool import (
+    close_database_connection_pool,
+    get_database_engine,
+    is_memory_sqlite,
+)
 from ..server.core import NoEntry
 from ..server.schemas import Asset, DataSource, Management, Revision
 from ..server.settings import DatabaseSettings
@@ -229,10 +233,7 @@ async def execute(self, statement, explain=None):
             return result
 
     async def startup(self):
-        if (self.engine.dialect.name == "sqlite") and (
-            self.engine.url.database == ":memory:"
-            or self.engine.url.query.get("mode") == "memory"
-        ):
+        if is_memory_sqlite(self.engine.url):
             # Special-case for in-memory SQLite: Because it is transient we can
             # skip over anything related to migrations.
             await initialize_database(self.engine)

tiled/server/connection_pool.py

@@ -5,7 +5,7 @@
 
 from fastapi import Depends
 from sqlalchemy import event
-from sqlalchemy.engine import make_url
+from sqlalchemy.engine import URL, make_url
 from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine
 from sqlalchemy.pool import AsyncAdaptedQueuePool
 
@@ -55,20 +55,14 @@ async def __aexit__(self, *excinfo):
 
 
 def open_database_connection_pool(database_settings: DatabaseSettings) -> AsyncEngine:
-    if (
-        make_url(database_settings.uri).database == ":memory:"
-        or database_settings.uri == "sqlite://"
-    ):
-        # For SQLite databases that exist only in process memory,
+    if is_memory_sqlite(database_settings.uri):
         engine = create_async_engine(
             ensure_specified_sql_driver(database_settings.uri),
             echo=DEFAULT_ECHO,
             json_serializer=json_serializer,
         )
 
     else:
-        # For file-backed SQLite databases, and for PostgreSQL databases,
-        # connection pooling offers a significant performance boost.
         engine = create_async_engine(
             ensure_specified_sql_driver(database_settings.uri),
             echo=DEFAULT_ECHO,
@@ -139,3 +133,30 @@ def _set_sqlite_pragma(conn, record):
     cursor = conn.cursor()
     cursor.execute("PRAGMA foreign_keys=ON")
     cursor.close()
+
+
+def is_memory_sqlite(url: Union[URL, str]) -> bool:
+    """
+    Check if a SQLAlchemy URL is a memory-backed SQLite database.
+
+    Handles various memory database URL formats:
+    - sqlite:///:memory:
+    - sqlite:///file::memory:?cache=shared
+    - sqlite://
+    - etc.
+    """
+    url = make_url(url)
+    # Check if it's SQLite at all
+    if url.get_dialect().name != "sqlite":
+        return False
+
+    # Check if database is None or empty (default memory DB)
+    if not url.database:
+        return True
+
+    # Check for explicit :memory: string (case-insensitive)
+    database = str(url.database).lower()
+    if ":memory:" in database:
+        return True
+
+    return False