Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 18 additions & 1 deletion pkgs/standards/autoapi/autoapi/v3/ddl/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
22 changes: 22 additions & 0 deletions pkgs/standards/autoapi/tests/unit/test_sqlite_attachments.py
Original file line number Diff line number Diff line change
@@ -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):
Expand Down Expand Up @@ -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()
Loading