Skip to content
Merged
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
20 changes: 14 additions & 6 deletions packages/core/src/repowise/core/persistence/database.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
from pathlib import Path

import structlog
from sqlalchemy import event, inspect
from sqlalchemy import event, inspect, literal
from sqlalchemy.ext.asyncio import (
AsyncEngine,
AsyncSession,
Expand All @@ -44,6 +44,7 @@
# for large repos. SQLite blocks (doesn't busy-loop) so this is cheap.
_SQLITE_BUSY_TIMEOUT_MS = 30000


def _sqlite_pragmas(busy_timeout_ms: int) -> tuple[tuple[str, str], ...]:
"""Return the pragma list to apply to a SQLite connection.

Expand Down Expand Up @@ -299,7 +300,7 @@ async def get_session(
raise


def _column_default_sql(column: object) -> str | None:
def _column_default_sql(column: object, dialect: object) -> str | None:
"""Return a SQL literal/expression suitable for an ADD COLUMN DEFAULT.

Prefers ``server_default`` (the DDL-level default that the migration
Expand All @@ -323,7 +324,15 @@ def _column_default_sql(column: object) -> str | None:
arg = getattr(py_default, "arg", None)
if arg is not None and not callable(arg):
if isinstance(arg, bool):
return "1" if arg else "0"
# Boolean literals are dialect-specific: SQLite accepts 0/1,
# while PostgreSQL requires false/true for BOOLEAN columns.
# Compile the typed value instead of treating bool as int.
return str(
literal(arg, type_=column.type).compile( # type: ignore[attr-defined]
dialect=dialect,
compile_kwargs={"literal_binds": True},
)
)
if isinstance(arg, (int, float)):
return str(arg)
if isinstance(arg, str):
Expand All @@ -346,7 +355,7 @@ def _add_column_ddl(column: object, dialect: object) -> str:
f'"{column.name}"', # type: ignore[attr-defined]
column.type.compile(dialect=dialect), # type: ignore[attr-defined]
]
default_sql = _column_default_sql(column)
default_sql = _column_default_sql(column, dialect)
if default_sql is not None:
parts.append(f"DEFAULT {default_sql}")
if not column.nullable: # type: ignore[attr-defined]
Expand Down Expand Up @@ -447,8 +456,7 @@ def _run(what: str, build: Callable[[], object]) -> None:
_run(
f"{table.name}.{column.name}",
lambda table=table, column=column: text(
f'ALTER TABLE "{table.name}" ADD COLUMN '
f"{_add_column_ddl(column, dialect)}"
f'ALTER TABLE "{table.name}" ADD COLUMN {_add_column_ddl(column, dialect)}'
),
)

Expand Down
24 changes: 24 additions & 0 deletions tests/unit/persistence/test_schema_reconciliation.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,11 +17,35 @@
from typing import Any

import pytest
from sqlalchemy import Boolean, Column
from sqlalchemy.dialects import postgresql, sqlite

from repowise.core.persistence import create_engine, init_db
from repowise.core.persistence.models import Base


@pytest.mark.parametrize(
("dialect", "value", "expected"),
[
(sqlite.dialect(), False, '"pinned" BOOLEAN DEFAULT 0 NOT NULL'),
(sqlite.dialect(), True, '"pinned" BOOLEAN DEFAULT 1 NOT NULL'),
(postgresql.dialect(), False, '"pinned" BOOLEAN DEFAULT false NOT NULL'),
(postgresql.dialect(), True, '"pinned" BOOLEAN DEFAULT true NOT NULL'),
],
)
def test_python_boolean_defaults_use_dialect_literals(
dialect: object,
value: bool,
expected: str,
) -> None:
"""Legacy-column DDL must be accepted by both supported databases."""
from repowise.core.persistence.database import _add_column_ddl

column = Column("pinned", Boolean, nullable=False, default=value)

assert _add_column_ddl(column, dialect) == expected


def _table_columns(db_path: Path, table: str) -> set[str]:
conn = sqlite3.connect(db_path)
try:
Expand Down
Loading