diff --git a/pkgs/standards/autoapi/autoapi/v3/ddl/__init__.py b/pkgs/standards/autoapi/autoapi/v3/ddl/__init__.py index fc0bd6a9bb..d5ce5b271b 100644 --- a/pkgs/standards/autoapi/autoapi/v3/ddl/__init__.py +++ b/pkgs/standards/autoapi/autoapi/v3/ddl/__init__.py @@ -209,12 +209,29 @@ def _create_all_on_bind( attachments = sqlite_default_attach_map(engine, schema_names) if attachments: - bootstrap_dbschema( + info = bootstrap_dbschema( engine, schemas=schema_names, sqlite_attachments=attachments, immediate=True, ) + # If schemas were collected, treat the DSN path as a base path by + # removing the original SQLite file once schema-specific attachments + # have been created. This avoids leaving behind an unused ``main`` + # database like ``authn.db`` alongside the expected + # ``authn__authn.db`` attachment. + if getattr(engine.dialect, "name", "") == "sqlite": + base_db = getattr(getattr(engine, "url", None), "database", None) + if base_db and base_db not in (":memory:", ""): + base_path = Path(base_db) + attached_paths = { + Path(p) for p in info.get("sqlite_attachments", {}).values() + } + if base_path.exists() and base_path not in attached_paths: + try: + base_path.unlink() + except OSError: + pass else: ensure_schemas(engine, schema_names) diff --git a/pkgs/standards/autoapi/tests/unit/test_sqlite_attachments.py b/pkgs/standards/autoapi/tests/unit/test_sqlite_attachments.py index 64af776267..7ce2a05c7f 100644 --- a/pkgs/standards/autoapi/tests/unit/test_sqlite_attachments.py +++ b/pkgs/standards/autoapi/tests/unit/test_sqlite_attachments.py @@ -1,5 +1,8 @@ import pytest from autoapi.v3.autoapp import AutoApp +from autoapi.v3.orm.tables import Base +from autoapi.v3.orm.mixins import GUIDPk +from autoapi.v3.types import Column, String def _db_names(conn): @@ -47,3 +50,22 @@ async def test_initialize_async_with_sqlite_attachments(async_db_session, tmp_pa result = await conn.exec_driver_sql("PRAGMA database_list") names = {row[1] for row in result.fetchall()} assert "logs" in names + + +@pytest.mark.asyncio +async def test_dsn_treated_as_base_path(tmp_path): + """Ensure only schema-attached SQLite database files are created.""" + base = tmp_path / "authn.db" + + class Model(Base, GUIDPk): + __tablename__ = "things" + __table_args__ = {"schema": "authn"} + name = Column(String, nullable=False) + + api = AutoApp(engine=f"sqlite+aiosqlite:///{base}") + api.include_model(Model) + await api.initialize() + + attach_path = tmp_path / "authn__authn.db" + assert attach_path.exists() + assert not base.exists()