Feat/datainfra refactor tracker

alembic/versions/0007_paper_harvest_tables.py

@@ -42,12 +42,21 @@ def _get_indexes(table: str) -> set[str]:
     return idx
 
 
+def _get_columns(table: str) -> set[str]:
+    cols = set()
+    for c in _insp().get_columns(table):
+        cols.add(str(c.get("name") or ""))
+    return cols
+
+
 def _create_index(name: str, table: str, cols: list[str]) -> None:
     if _is_offline():
         op.create_index(name, table, cols)
         return
     if name in _get_indexes(table):
         return
+    if not set(cols).issubset(_get_columns(table)):
+        return
     op.create_index(name, table, cols)
 
 

alembic/versions/0010_contract_feedback_fk.py

@@ -0,0 +1,97 @@
+"""contract feedback fk read path
+
+Revision ID: 0010_contract_feedback_fk
+Revises: 0009_paper_identifiers
+Create Date: 2026-02-12 08:05:00
+
+Contract phase for canonical feedback join:
+- Backfill canonical_paper_id from paper_ref_id when available.
+- Add index optimized for library reads by canonical FK.
+- Drop legacy paper_id index used by external-id joins.
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import context, op
+
+
+revision = "0010_contract_feedback_fk"
+down_revision = "0009_paper_identifiers"
+branch_labels = None
+depends_on = None
+
+
+def _is_offline() -> bool:
+    try:
+        return bool(context.is_offline_mode())
+    except Exception:
+        return False
+
+
+def _inspector():
+    bind = op.get_bind()
+    return sa.inspect(bind)
+
+
+def _has_table(name: str) -> bool:
+    if _is_offline():
+        return False
+    return bool(_inspector().has_table(name))
+
+
+def _get_columns(table: str) -> set[str]:
+    if _is_offline() or not _has_table(table):
+        return set()
+    return {c["name"] for c in _inspector().get_columns(table)}
+
+
+def _has_index(table: str, index_name: str) -> bool:
+    if _is_offline() or not _has_table(table):
+        return False
+    names = {idx.get("name") for idx in _inspector().get_indexes(table)}
+    return index_name in names
+
+
+def _create_index(name: str, table: str, cols: list[str]) -> None:
+    if _is_offline() or _has_index(table, name):
+        return
+    op.create_index(name, table, cols)
+
+
+def _drop_index(name: str, table: str) -> None:
+    if _is_offline() or not _has_index(table, name):
+        return
+    op.drop_index(name, table_name=table)
+
+
+def upgrade() -> None:
+    if not _has_table("paper_feedback"):
+        return
+
+    cols = _get_columns("paper_feedback")
+    if {"canonical_paper_id", "paper_ref_id"}.issubset(cols):
+        op.execute(
+            sa.text(
+                """
+                UPDATE paper_feedback
+                SET canonical_paper_id = paper_ref_id
+                WHERE canonical_paper_id IS NULL
+                  AND paper_ref_id IS NOT NULL
+                """
+            )
+        )
+
+    _create_index(
+        "ix_paper_feedback_user_action_canonical",
+        "paper_feedback",
+        ["user_id", "action", "canonical_paper_id"],
+    )
+
+    # Legacy external-id join path index.
+    _drop_index("ix_paper_feedback_paper_id", "paper_feedback")
+
+
+def downgrade() -> None:
+    _create_index("ix_paper_feedback_paper_id", "paper_feedback", ["paper_id"])
+    _drop_index("ix_paper_feedback_user_action_canonical", "paper_feedback")

alembic/versions/0011_model_endpoints.py

@@ -0,0 +1,72 @@
+"""add model endpoint gateway table
+
+Revision ID: 0011_model_endpoints
+Revises: 0010_contract_feedback_fk
+Create Date: 2026-02-12 11:10:00
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import context, op
+
+
+revision = "0011_model_endpoints"
+down_revision = "0010_contract_feedback_fk"
+branch_labels = None
+depends_on = None
+
+
+def _is_offline() -> bool:
+    try:
+        return bool(context.is_offline_mode())
+    except Exception:
+        return False
+
+
+def _inspector():
+    bind = op.get_bind()
+    return sa.inspect(bind)
+
+
+def _has_table(name: str) -> bool:
+    if _is_offline():
+        return False
+    return bool(_inspector().has_table(name))
+
+
+def upgrade() -> None:
+    if _has_table("model_endpoints"):
+        return
+
+    op.create_table(
+        "model_endpoints",
+        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column("name", sa.String(length=64), nullable=False),
+        sa.Column(
+            "vendor", sa.String(length=32), nullable=False, server_default="openai_compatible"
+        ),
+        sa.Column("base_url", sa.String(length=512), nullable=True),
+        sa.Column(
+            "api_key_env", sa.String(length=64), nullable=False, server_default="OPENAI_API_KEY"
+        ),
+        sa.Column("models_json", sa.Text(), nullable=False, server_default="[]"),
+        sa.Column("task_types_json", sa.Text(), nullable=False, server_default="[]"),
+        sa.Column("enabled", sa.Boolean(), nullable=False, server_default=sa.text("1")),
+        sa.Column("is_default", sa.Boolean(), nullable=False, server_default=sa.text("0")),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.UniqueConstraint("name", name="uq_model_endpoints_name"),
+    )
+    op.create_index("ix_model_endpoints_name", "model_endpoints", ["name"])
+    op.create_index("ix_model_endpoints_enabled", "model_endpoints", ["enabled"])
+    op.create_index("ix_model_endpoints_is_default", "model_endpoints", ["is_default"])
+
+
+def downgrade() -> None:
+    if not _has_table("model_endpoints"):
+        return
+    op.drop_index("ix_model_endpoints_is_default", table_name="model_endpoints")
+    op.drop_index("ix_model_endpoints_enabled", table_name="model_endpoints")
+    op.drop_index("ix_model_endpoints_name", table_name="model_endpoints")
+    op.drop_table("model_endpoints")

alembic/versions/0012_reconcile_papers_schema.py

@@ -0,0 +1,134 @@
+"""reconcile papers schema with current ORM
+
+Revision ID: 0012_reconcile_papers_schema
+Revises: 0011_model_endpoints
+Create Date: 2026-02-12 12:20:00
+
+Adds missing columns/indexes on legacy `papers` table created before 0007.
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import context, op
+
+
+revision = "0012_reconcile_papers_schema"
+down_revision = "0011_model_endpoints"
+branch_labels = None
+depends_on = None
+
+
+def _is_offline() -> bool:
+    try:
+        return bool(context.is_offline_mode())
+    except Exception:
+        return False
+
+
+def _inspector():
+    return sa.inspect(op.get_bind())
+
+
+def _has_table(name: str) -> bool:
+    if _is_offline():
+        return False
+    return bool(_inspector().has_table(name))
+
+
+def _columns(table: str) -> set[str]:
+    if _is_offline() or not _has_table(table):
+        return set()
+    return {str(c.get("name") or "") for c in _inspector().get_columns(table)}
+
+
+def _has_index(table: str, index_name: str) -> bool:
+    if _is_offline() or not _has_table(table):
+        return False
+    names = {str(i.get("name") or "") for i in _inspector().get_indexes(table)}
+    return index_name in names
+
+
+def _add_column_if_missing(table: str, column: sa.Column) -> None:
+    if _is_offline() or column.name in _columns(table):
+        return
+    op.add_column(table, column)
+
+
+def _create_index_if_possible(name: str, table: str, cols: list[str]) -> None:
+    if _is_offline() or _has_index(table, name):
+        return
+    if not set(cols).issubset(_columns(table)):
+        return
+    op.create_index(name, table, cols)
+
+
+def upgrade() -> None:
+    if not _has_table("papers"):
+        return
+
+    _add_column_if_missing(
+        "papers", sa.Column("semantic_scholar_id", sa.String(length=64), nullable=True)
+    )
+    _add_column_if_missing("papers", sa.Column("openalex_id", sa.String(length=64), nullable=True))
+    _add_column_if_missing(
+        "papers",
+        sa.Column("title_hash", sa.String(length=64), nullable=False, server_default=""),
+    )
+    _add_column_if_missing("papers", sa.Column("year", sa.Integer(), nullable=True))
+    _add_column_if_missing(
+        "papers", sa.Column("publication_date", sa.String(length=32), nullable=True)
+    )
+    _add_column_if_missing(
+        "papers", sa.Column("citation_count", sa.Integer(), nullable=False, server_default="0")
+    )
+    _add_column_if_missing(
+        "papers",
+        sa.Column("fields_of_study_json", sa.Text(), nullable=False, server_default="[]"),
+    )
+    _add_column_if_missing(
+        "papers",
+        sa.Column("primary_source", sa.String(length=32), nullable=False, server_default=""),
+    )
+    _add_column_if_missing(
+        "papers",
+        sa.Column("sources_json", sa.Text(), nullable=False, server_default="[]"),
+    )
+    _add_column_if_missing(
+        "papers", sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True)
+    )
+
+    # Fill empty title_hash so ORM non-null assumptions hold.
+    if "title_hash" in _columns("papers"):
+        op.execute(
+            sa.text(
+                """
+                UPDATE papers
+                SET title_hash = lower(hex(randomblob(16)))
+                WHERE title_hash IS NULL OR title_hash = ''
+                """
+            )
+        )
+
+    _create_index_if_possible("ix_papers_semantic_scholar_id", "papers", ["semantic_scholar_id"])
+    _create_index_if_possible("ix_papers_openalex_id", "papers", ["openalex_id"])
+    _create_index_if_possible("ix_papers_title_hash", "papers", ["title_hash"])
+    _create_index_if_possible("ix_papers_year", "papers", ["year"])
+    _create_index_if_possible("ix_papers_venue", "papers", ["venue"])
+    _create_index_if_possible("ix_papers_citation_count", "papers", ["citation_count"])
+    _create_index_if_possible("ix_papers_primary_source", "papers", ["primary_source"])
+
+
+def downgrade() -> None:
+    # Keep columns for backward compatibility; only drop indexes created here.
+    for idx in [
+        "ix_papers_semantic_scholar_id",
+        "ix_papers_openalex_id",
+        "ix_papers_title_hash",
+        "ix_papers_year",
+        "ix_papers_venue",
+        "ix_papers_citation_count",
+        "ix_papers_primary_source",
+    ]:
+        if _has_index("papers", idx):
+            op.drop_index(idx, table_name="papers")

alembic/versions/0013_model_endpoint_api_key.py

@@ -0,0 +1,56 @@
+"""add api key storage for model endpoints
+
+Revision ID: 0013_model_endpoint_api_key
+Revises: 0012_reconcile_papers_schema
+Create Date: 2026-02-12 20:30:00
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import context, op
+
+
+revision = "0013_model_endpoint_api_key"
+down_revision = "0012_reconcile_papers_schema"
+branch_labels = None
+depends_on = None
+
+
+def _is_offline() -> bool:
+    try:
+        return bool(context.is_offline_mode())
+    except Exception:
+        return False
+
+
+def _inspector():
+    return sa.inspect(op.get_bind())
+
+
+def _has_table(name: str) -> bool:
+    if _is_offline():
+        return False
+    return bool(_inspector().has_table(name))
+
+
+def _columns(table: str) -> set[str]:
+    if _is_offline() or not _has_table(table):
+        return set()
+    return {str(c.get("name") or "") for c in _inspector().get_columns(table)}
+
+
+def upgrade() -> None:
+    if not _has_table("model_endpoints"):
+        return
+    if "api_key_value" not in _columns("model_endpoints"):
+        op.add_column(
+            "model_endpoints", sa.Column("api_key_value", sa.String(length=512), nullable=True)
+        )
+
+
+def downgrade() -> None:
+    if not _has_table("model_endpoints"):
+        return
+    if "api_key_value" in _columns("model_endpoints"):
+        op.drop_column("model_endpoints", "api_key_value")