feat: store matching keys and duplicate dismissals
Derive normalized_title, normalized_name and normalized_value with @validates so no write can bypass them, and add the table recording pairs a reader has said are not the same book.
This commit is contained in:
+167
@@ -0,0 +1,167 @@
|
||||
"""add book matching keys and duplicate dismissals
|
||||
|
||||
Revision ID: 4358e7d4743a
|
||||
Revises: e9c2c7e875ae
|
||||
Create Date: 2026-08-15 15:09:02.708914
|
||||
|
||||
"""
|
||||
|
||||
import warnings
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
from advanced_alchemy.types import EncryptedString, EncryptedText, GUID, ORA_JSONB, DateTimeUTC, StoredObject, PasswordHash, FernetBackend
|
||||
from advanced_alchemy.types.encrypted_string import PGCryptoBackend
|
||||
from advanced_alchemy.types.password_hash.argon2 import Argon2Hasher
|
||||
from advanced_alchemy.types.password_hash.passlib import PasslibHasher
|
||||
from advanced_alchemy.types.password_hash.pwdlib import PwdlibHasher
|
||||
from sqlalchemy import Text # noqa: F401
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Sequence
|
||||
|
||||
__all__ = ["downgrade", "upgrade", "schema_upgrades", "schema_downgrades", "data_upgrades", "data_downgrades"]
|
||||
|
||||
sa.GUID = GUID
|
||||
sa.DateTimeUTC = DateTimeUTC
|
||||
sa.ORA_JSONB = ORA_JSONB
|
||||
sa.EncryptedString = EncryptedString
|
||||
sa.EncryptedText = EncryptedText
|
||||
sa.StoredObject = StoredObject
|
||||
sa.PasswordHash = PasswordHash
|
||||
sa.Argon2Hasher = Argon2Hasher
|
||||
sa.PasslibHasher = PasslibHasher
|
||||
sa.PwdlibHasher = PwdlibHasher
|
||||
sa.FernetBackend = FernetBackend
|
||||
sa.PGCryptoBackend = PGCryptoBackend
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision = '4358e7d4743a'
|
||||
down_revision = 'e9c2c7e875ae'
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
with warnings.catch_warnings():
|
||||
warnings.filterwarnings("ignore", category=UserWarning)
|
||||
with op.get_context().autocommit_block():
|
||||
schema_upgrades()
|
||||
data_upgrades()
|
||||
|
||||
def downgrade() -> None:
|
||||
with warnings.catch_warnings():
|
||||
warnings.filterwarnings("ignore", category=UserWarning)
|
||||
with op.get_context().autocommit_block():
|
||||
data_downgrades()
|
||||
schema_downgrades()
|
||||
|
||||
def schema_upgrades() -> None:
|
||||
"""schema upgrade migrations go here."""
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
op.create_table('duplicate_dismissals',
|
||||
sa.Column('id', sa.BigInteger().with_variant(sa.Integer(), 'sqlite'), nullable=False),
|
||||
sa.Column('book_a_id', sa.BigInteger().with_variant(sa.Integer(), 'sqlite'), nullable=False),
|
||||
sa.Column('book_b_id', sa.BigInteger().with_variant(sa.Integer(), 'sqlite'), nullable=False),
|
||||
sa.ForeignKeyConstraint(['book_a_id'], ['books.id'], name=op.f('fk_duplicate_dismissals_book_a_id_books'), ondelete='cascade'),
|
||||
sa.ForeignKeyConstraint(['book_b_id'], ['books.id'], name=op.f('fk_duplicate_dismissals_book_b_id_books'), ondelete='cascade'),
|
||||
sa.PrimaryKeyConstraint('id', name=op.f('pk_duplicate_dismissals')),
|
||||
sa.UniqueConstraint('book_a_id', 'book_b_id', name=op.f('uq_duplicate_dismissals_book_a_id'))
|
||||
)
|
||||
with op.batch_alter_table('duplicate_dismissals', schema=None) as batch_op:
|
||||
batch_op.create_index(batch_op.f('ix_duplicate_dismissals_book_a_id'), ['book_a_id'], unique=False)
|
||||
batch_op.create_index(batch_op.f('ix_duplicate_dismissals_book_b_id'), ['book_b_id'], unique=False)
|
||||
|
||||
# `server_default` so the column can be added to a table that already has rows;
|
||||
# `data_upgrades` fills in the real keys immediately afterwards.
|
||||
with op.batch_alter_table('authors', schema=None) as batch_op:
|
||||
batch_op.add_column(sa.Column('normalized_name', sa.String(), nullable=False, server_default=''))
|
||||
batch_op.create_index(batch_op.f('ix_authors_normalized_name'), ['normalized_name'], unique=False)
|
||||
|
||||
with op.batch_alter_table('books', schema=None) as batch_op:
|
||||
batch_op.add_column(sa.Column('normalized_title', sa.String(), nullable=False, server_default=''))
|
||||
batch_op.create_index(batch_op.f('ix_books_normalized_title'), ['normalized_title'], unique=False)
|
||||
|
||||
with op.batch_alter_table('identifiers', schema=None) as batch_op:
|
||||
batch_op.add_column(sa.Column('normalized_value', sa.String(), nullable=True))
|
||||
batch_op.create_index(batch_op.f('ix_identifiers_normalized_value'), ['normalized_value'], unique=False)
|
||||
|
||||
# ### end Alembic commands ###
|
||||
|
||||
def schema_downgrades() -> None:
|
||||
"""schema downgrade migrations go here."""
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
with op.batch_alter_table('identifiers', schema=None) as batch_op:
|
||||
batch_op.drop_index(batch_op.f('ix_identifiers_normalized_value'))
|
||||
batch_op.drop_column('normalized_value')
|
||||
|
||||
with op.batch_alter_table('books', schema=None) as batch_op:
|
||||
batch_op.drop_index(batch_op.f('ix_books_normalized_title'))
|
||||
batch_op.drop_column('normalized_title')
|
||||
|
||||
with op.batch_alter_table('authors', schema=None) as batch_op:
|
||||
batch_op.drop_index(batch_op.f('ix_authors_normalized_name'))
|
||||
batch_op.drop_column('normalized_name')
|
||||
|
||||
with op.batch_alter_table('duplicate_dismissals', schema=None) as batch_op:
|
||||
batch_op.drop_index(batch_op.f('ix_duplicate_dismissals_book_b_id'))
|
||||
batch_op.drop_index(batch_op.f('ix_duplicate_dismissals_book_a_id'))
|
||||
|
||||
op.drop_table('duplicate_dismissals')
|
||||
# ### end Alembic commands ###
|
||||
|
||||
def data_upgrades() -> None:
|
||||
"""
|
||||
Fill the matching keys in for rows that already exist.
|
||||
|
||||
The validators on the models only fire when something is written, so without this
|
||||
every book imported before today is invisible to duplicate detection. Run through
|
||||
the same helpers the validators use, so a backfilled row and a freshly written one
|
||||
are guaranteed to agree.
|
||||
"""
|
||||
from chitai.services.matching import (
|
||||
normalize_author,
|
||||
normalize_identifier,
|
||||
normalize_title,
|
||||
)
|
||||
|
||||
connection = op.get_bind()
|
||||
|
||||
books = connection.execute(sa.text("SELECT id, title FROM books")).fetchall()
|
||||
_apply(
|
||||
connection,
|
||||
"UPDATE books SET normalized_title = :key WHERE id = :id",
|
||||
[{"id": id, "key": normalize_title(title)} for id, title in books],
|
||||
)
|
||||
|
||||
authors = connection.execute(sa.text("SELECT id, name FROM authors")).fetchall()
|
||||
_apply(
|
||||
connection,
|
||||
"UPDATE authors SET normalized_name = :key WHERE id = :id",
|
||||
[{"id": id, "key": normalize_author(name)} for id, name in authors],
|
||||
)
|
||||
|
||||
identifiers = connection.execute(
|
||||
sa.text("SELECT id, name, value FROM identifiers")
|
||||
).fetchall()
|
||||
_apply(
|
||||
connection,
|
||||
"UPDATE identifiers SET normalized_value = :key WHERE id = :id",
|
||||
[
|
||||
{"id": id, "key": normalize_identifier(name, value)}
|
||||
for id, name, value in identifiers
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def _apply(connection, statement: str, parameters: list[dict]) -> None:
|
||||
"""Run one update per row, in batches, skipping the work when there are none."""
|
||||
batch_size = 1000
|
||||
|
||||
for start in range(0, len(parameters), batch_size):
|
||||
connection.execute(sa.text(statement), parameters[start : start + batch_size])
|
||||
|
||||
|
||||
def data_downgrades() -> None:
|
||||
"""Add any optional data downgrade migrations here!"""
|
||||
Reference in New Issue
Block a user