feat: unify database_url config for Postgres and SQLite

Extend BASIC_MEMORY_DATABASE_URL to support:
- SQLite URLs with custom paths for project-local databases
- Postgres URLs with ?search_path= for schema isolation

Changes:
- config.py: Parse SQLite URLs in app_database_path property
- db.py: Extract search_path from Postgres URL, pass to server_settings
- db.py: Add ensure_schema_exists() for auto-creating schemas
- alembic/env.py: Pass version_table_schema for migration tracking
- cli/commands/db.py: Handle Postgres case in reset command

Closes basicmachines-co#539

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Signed-off-by: Cedric Hurst <cedric@spantree.net>

Author: divideby0

src/basic_memory/alembic/env.py

@@ -98,14 +98,50 @@ def run_migrations_offline() -> None:
         context.run_migrations()
 
 
+def get_version_table_schema(connection) -> str | None:
+    """Determine the schema for Alembic's version table.
+
+    Args:
+        connection: Database connection
+
+    Returns:
+        Schema name if Postgres with non-public search_path, else None
+
+    Why: When using schema isolation, Alembic's alembic_version table should
+    be in the same schema as the application tables.
+    """
+    if connection.dialect.name != "postgresql":
+        return None
+
+    # Check if database_url has search_path parameter
+    if not app_config.database_url or "search_path=" not in app_config.database_url:
+        return None
+
+    from urllib.parse import urlparse, parse_qs
+
+    parsed = urlparse(app_config.database_url)
+    query_params = parse_qs(parsed.query)
+    search_path = query_params.get("search_path", ["public"])[0]
+
+    # Only set version_table_schema for non-public schemas
+    return search_path if search_path != "public" else None
+
+
 def do_run_migrations(connection):
     """Execute migrations with the given connection."""
+    # --- Schema-Aware Migration Tracking ---
+    # Trigger: Postgres with non-public search_path in database_url
+    # Why: Alembic version table should be in same schema as application tables
+    # Outcome: version_table_schema passed to context.configure()
+    version_table_schema = get_version_table_schema(connection)
+
     context.configure(
         connection=connection,
         target_metadata=target_metadata,
         include_object=include_object,
         render_as_batch=True,
         compare_type=True,
+        version_table_schema=version_table_schema,
     )
     with context.begin_transaction():
         context.run_migrations()

src/basic_memory/cli/commands/db.py

@@ -61,23 +61,34 @@ def reset(
         logger.info("Resetting database...")
         config_manager = ConfigManager()
         app_config = config_manager.config
-        # Get database path
+        # Get database path (None for Postgres)
         db_path = app_config.app_database_path
 
-        # Delete the database file and WAL files if they exist
-        for suffix in ["", "-shm", "-wal"]:
-            path = db_path.parent / f"{db_path.name}{suffix}"
-            if path.exists():
-                try:
-                    path.unlink()
-                    logger.info(f"Deleted: {path}")
-                except OSError as e:
-                    console.print(
-                        f"[red]Error:[/red] Cannot delete {path.name}: {e}\n"
-                        "The database may be in use by another process (e.g., MCP server).\n"
-                        "Please close Claude Desktop or any other Basic Memory clients and try again."
-                    )
-                    raise typer.Exit(1)
+        # --- SQLite vs Postgres Handling ---
+        # Trigger: db_path is None when using Postgres backend
+        # Why: Postgres doesn't use a local file, so file deletion doesn't apply
+        # Outcome: skip file deletion for Postgres, only run migrations
+        if db_path is not None:
+            # Delete the database file and WAL files if they exist (SQLite only)
+            for suffix in ["", "-shm", "-wal"]:
+                path = db_path.parent / f"{db_path.name}{suffix}"
+                if path.exists():
+                    try:
+                        path.unlink()
+                        logger.info(f"Deleted: {path}")
+                    except OSError as e:
+                        console.print(
+                            f"[red]Error:[/red] Cannot delete {path.name}: {e}\n"
+                            "The database may be in use by another process (e.g., MCP server).\n"
+                            "Please close Claude Desktop or any other Basic Memory clients and try again."
+                        )
+                        raise typer.Exit(1)
+        else:
+            console.print(
+                "[yellow]Note:[/yellow] Using Postgres backend. "
+                "Database files cannot be deleted directly.\n"
+                "Running migrations to recreate tables..."
+            )
 
         # Create a new empty database (preserves project configuration)
         try:

src/basic_memory/config.py

@@ -316,24 +316,49 @@ def model_post_init(self, __context: Any) -> None:
             self.default_project = next(iter(self.projects.keys()))
 
     @property
-    def app_database_path(self) -> Path:
+    def app_database_path(self) -> Optional[Path]:
         """Get the path to the app-level database.
 
         This is the single database that will store all knowledge data
         across all projects.
+
+        Returns:
+            Path to SQLite database file, or None for Postgres backends.
         """
+        # --- SQLite URL Handling ---
+        # Trigger: database_url is set and starts with "sqlite"
+        # Why: allows project-local SQLite databases via URL configuration
+        # Outcome: extracts path from URL (e.g., sqlite+aiosqlite:///.basic-memory/memory.db)
+        if self.database_url:
+            if self.database_url.startswith("sqlite"):
+                from urllib.parse import urlparse
+
+                parsed = urlparse(self.database_url)
+                # parsed.path will be "/.basic-memory/memory.db" - strip leading /
+                path = parsed.path[1:] if parsed.path.startswith("/") else parsed.path
+                resolved_path = Path(path).resolve()
+                # Ensure parent directory exists
+                if not resolved_path.parent.exists():  # pragma: no cover
+                    resolved_path.parent.mkdir(parents=True, exist_ok=True)
+                return resolved_path
+            else:
+                # Postgres or other backend - no file path
+                return None
+
+        # --- Default SQLite Path ---
+        # No database_url set - use default ~/.basic-memory/memory.db
         database_path = Path.home() / DATA_DIR_NAME / APP_DATABASE_NAME
         if not database_path.exists():  # pragma: no cover
             database_path.parent.mkdir(parents=True, exist_ok=True)
             database_path.touch()
         return database_path
 
     @property
-    def database_path(self) -> Path:
+    def database_path(self) -> Optional[Path]:
         """Get SQLite database path.
 
-        Rreturns the app-level database path
-        for backward compatibility in the codebase.
+        Returns the app-level database path for backward compatibility.
+        Returns None when using Postgres backend.
         """
 
         # Load the app-level database path from the global config

src/basic_memory/db.py

@@ -83,7 +83,15 @@ def get_db_url(
             logger.info(f"Using Postgres database: {config.database_url}")
             return config.database_url
 
-        # SQLite databases
+        # --- SQLite URL Handling ---
+        # Trigger: database_url is set with a SQLite URL
+        # Why: allows custom SQLite paths via URL configuration
+        # Outcome: use the provided URL instead of constructing from db_path
+        if config.database_url and config.database_url.startswith("sqlite"):
+            logger.info(f"Using SQLite database from URL: {config.database_url}")
+            return config.database_url
+
+        # SQLite databases (default behavior)
         if db_type == cls.MEMORY:
             logger.info("Using in-memory SQLite database")
             return "sqlite+aiosqlite://"
@@ -206,6 +214,34 @@ def enable_wal_mode(dbapi_conn, connection_record):
     return engine
 
 
+def extract_search_path_from_url(db_url: str) -> tuple[str, str]:
+    """Extract search_path from Postgres URL and return clean URL.
+
+    Args:
+        db_url: Postgres connection URL, possibly with ?search_path=schema
+
+    Returns:
+        Tuple of (clean_url without search_path, search_path value)
+
+    Why: asyncpg rejects search_path as a URL query parameter, so we extract it
+    and pass it via server_settings instead.
+    """
+    from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
+
+    parsed = urlparse(db_url)
+    query_params = parse_qs(parsed.query)
+
+    # Extract search_path, default to "public"
+    search_path_list = query_params.pop("search_path", ["public"])
+    search_path = search_path_list[0] if search_path_list else "public"
+
+    # Rebuild URL without search_path
+    new_query = urlencode(query_params, doseq=True)
+    clean_url = urlunparse(parsed._replace(query=new_query))
+
+    return clean_url, search_path
+
+
 def _create_postgres_engine(db_url: str, config: BasicMemoryConfig) -> AsyncEngine:
     """Create Postgres async engine with appropriate configuration.
 
@@ -216,10 +252,16 @@ def _create_postgres_engine(db_url: str, config: BasicMemoryConfig) -> AsyncEngi
     Returns:
         Configured async engine for Postgres
     """
+    # --- Extract search_path from URL ---
+    # Trigger: URL contains ?search_path=schema parameter
+    # Why: asyncpg rejects search_path as URL param, must pass via server_settings
+    # Outcome: clean URL for asyncpg, search_path passed to server_settings
+    clean_url, search_path = extract_search_path_from_url(db_url)
+
     # Use NullPool connection issues.
     # Assume connection pooler like PgBouncer handles connection pooling.
     engine = create_async_engine(
-        db_url,
+        clean_url,
         echo=False,
         poolclass=NullPool,  # No pooling - fresh connection per request
         connect_args={
@@ -233,10 +275,12 @@ def _create_postgres_engine(db_url: str, config: BasicMemoryConfig) -> AsyncEngi
                 "application_name": "basic-memory",
                 # Statement timeout for queries (30s to allow for cold start)
                 "statement_timeout": "30s",
+                # Schema isolation via search_path (extracted from URL or default "public")
+                "search_path": search_path,
             },
         },
     )
-    logger.debug("Created Postgres engine with NullPool (no connection pooling)")
+    logger.debug(f"Created Postgres engine with search_path={search_path}")
 
     return engine
 
@@ -365,6 +409,44 @@ async def engine_session_factory(
             _session_maker = None
 
 
+def get_search_path_from_config(app_config: BasicMemoryConfig) -> Optional[str]:
+    """Extract search_path from config's database_url if present.
+
+    Args:
+        app_config: BasicMemoryConfig with database_url
+
+    Returns:
+        search_path value if present and not "public", else None
+    """
+    if not app_config.database_url:
+        return None
+
+    if not app_config.database_url.startswith("postgresql"):
+        return None
+
+    _, search_path = extract_search_path_from_url(app_config.database_url)
+    return search_path if search_path != "public" else None
+
+
+async def ensure_schema_exists(engine: AsyncEngine, schema: str) -> None:
+    """Create schema if it doesn't exist (Postgres only).
+
+    Args:
+        engine: AsyncEngine connected to Postgres
+        schema: Schema name to create
+
+    Why: When using search_path for schema isolation, the schema must exist
+    before migrations can create tables in it.
+    """
+    if not schema or schema == "public":
+        return
+
+    async with engine.begin() as conn:
+        # Use text() to execute raw SQL - schema names are trusted config values
+        await conn.execute(text(f'CREATE SCHEMA IF NOT EXISTS "{schema}"'))
+    logger.info(f"Ensured schema exists: {schema}")
+
+
 async def run_migrations(
     app_config: BasicMemoryConfig, database_type=DatabaseType.FILESYSTEM
 ):  # pragma: no cover
@@ -393,6 +475,22 @@ async def run_migrations(
         db_url = DatabaseType.get_db_url(app_config.database_path, database_type, app_config)
         config.set_main_option("sqlalchemy.url", db_url)
 
+        # --- Schema Creation for Postgres ---
+        # Trigger: Postgres backend with non-public search_path in URL
+        # Why: schema must exist before Alembic can create tables in it
+        # Outcome: CREATE SCHEMA IF NOT EXISTS runs before migrations
+        search_path = get_search_path_from_config(app_config)
+        if search_path and (
+            database_type == DatabaseType.POSTGRES
+            or app_config.database_backend == DatabaseBackend.POSTGRES
+        ):
+            # Create a temporary engine just for schema creation
+            temp_engine = _create_postgres_engine(db_url, app_config)
+            try:
+                await ensure_schema_exists(temp_engine, search_path)
+            finally:
+                await temp_engine.dispose()
+
         command.upgrade(config, "head")
         logger.info("Migrations completed successfully")