restore bootstrap_simple_efp_dbs.py and efp_bootstrap.py needed for CI

rmobmina · claude · rmobmina · commit 16c51c247081 · 2026-03-12T10:26:09.000-04:00
These scripts are called by config/init.sh to create simple eFP MySQL databases
from schema definitions in CI, where SQLite mirrors are not available.

Co-Authored-By: Claude Sonnet 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/api/services/efp_bootstrap.py b/api/services/efp_bootstrap.py
@@ -0,0 +1,273 @@
+"""
+Utilities to bootstrap the simple eFP databases directly from the shared schema
+definitions. Shared by the CLI script and the Flask endpoint so we only maintain
+one implementation.
+"""
+
+from __future__ import annotations
+
+import re
+import hashlib
+from typing import Dict, Iterable, List
+
+from sqlalchemy import Column, Index, MetaData, Table, create_engine, text
+from sqlalchemy.dialects.mysql import FLOAT, INTEGER, TEXT, VARCHAR
+from sqlalchemy.engine import URL
+
+from api.models.efp_schemas import SIMPLE_EFP_DATABASE_SCHEMAS
+
+
+def _column_type(column_spec):
+    """
+    Convert a column specification dictionary to a SQLAlchemy MySQL column type.
+
+    :param column_spec: Dictionary containing column metadata from schema definition
+    :type column_spec: Dict[str, Any]
+    :return: SQLAlchemy column type object (VARCHAR, INTEGER, FLOAT, or TEXT)
+    :rtype: sqlalchemy.types.TypeEngine
+    :raises ValueError: If column type is not one of: string, integer, float, text
+    """
+    col_type = column_spec.get("type")
+    if col_type == "string":
+        return VARCHAR(column_spec["length"])
+    if col_type == "integer":
+        return INTEGER(unsigned=column_spec.get("unsigned", False))
+    if col_type == "float":
+        # explicit mysql float keeps parity with the original dumps
+        return FLOAT()
+    if col_type == "text":
+        return TEXT()
+    raise ValueError(f"Unsupported column type: {col_type}")
+
+
+def _build_table(metadata: MetaData, spec, db_name: str) -> Table:
+    """
+    Build a SQLAlchemy Table object from a schema specification.
+
+    Creates columns with proper types, constraints, defaults, and indexes based on
+    the schema definition. This Table object can be used to generate CREATE TABLE
+    SQL statements.
+
+    :param metadata: SQLAlchemy MetaData object to attach the table to
+    :type metadata: sqlalchemy.schema.MetaData
+    :param spec: Database schema specification from SIMPLE_EFP_DATABASE_SCHEMAS
+    :type spec: Dict[str, Any]
+    :param db_name: Name of the database (used for index naming)
+    :type db_name: str
+    :return: SQLAlchemy Table object with all columns and indexes defined
+    :rtype: sqlalchemy.schema.Table
+    """
+    columns = []
+    for column in spec["columns"]:
+        kwargs = {"nullable": column.get("nullable", True)}
+        if column.get("primary_key"):
+            kwargs["primary_key"] = True
+        default_value = column.get("default")
+        if default_value is not None:
+            if isinstance(default_value, str):
+                kwargs["server_default"] = text(f"'{default_value}'")
+            else:
+                kwargs["server_default"] = text(str(default_value))
+
+        columns.append(Column(column["name"], _column_type(column), **kwargs))
+
+    table = Table(spec["table_name"], metadata, *columns, mysql_charset=spec.get("charset"))
+    index_cols = spec.get("index") or []
+    if index_cols:
+        # MySQL cannot index TEXT/BLOB columns without a prefix length.
+        # Filter them out to avoid bootstrap failures when schemas use TEXT.
+        text_cols = {col["name"] for col in spec["columns"] if col.get("type") == "text"}
+        index_cols = [col for col in index_cols if col not in text_cols]
+    if index_cols:
+        index_name = _make_index_name(db_name, index_cols)
+        Index(index_name, *[table.c[col] for col in index_cols])
+    return table
+
+
+def _make_index_name(db_name: str, index_cols: Iterable[str], max_len: int = 64) -> str:
+    """
+    Create a MySQL-safe index name capped at 64 characters.
+
+    If the generated name is too long, fall back to a truncated db_name with a stable hash
+    to keep names deterministic and avoid collisions.
+    """
+    base = f"ix_{db_name}_{'_'.join(index_cols)}"
+    if len(base) <= max_len:
+        return base
+
+    digest = hashlib.sha1(base.encode("utf-8")).hexdigest()[:8]
+    reserved = len("ix_") + 1 + len(digest)
+    db_len = max_len - reserved
+    if db_len <= 0:
+        return f"ix_{digest}"
+    return f"ix_{db_name[:db_len]}_{digest}"
+
+
+def _build_url(host: str, port: int, user: str, password: str, database: str | None = None) -> URL:
+    """
+    Build a SQLAlchemy database URL for MySQL connections.
+
+    :param host: MySQL server hostname (e.g., 'localhost', 'BAR_mysqldb')
+    :type host: str
+    :param port: MySQL server port number (typically 3306)
+    :type port: int
+    :param user: MySQL username for authentication
+    :type user: str
+    :param password: MySQL password for authentication
+    :type password: str
+    :param database: Optional database name to connect to; if None, connects to server without selecting a database
+    :type database: str or None
+    :return: SQLAlchemy URL object for mysql+mysqldb connections
+    :rtype: sqlalchemy.engine.URL
+    """
+    return URL.create(
+        drivername="mysql+mysqldb",
+        username=user,
+        password=password,
+        host=host,
+        port=port,
+        database=database,
+    )
+
+
+def ensure_database(server_url: URL, db_name: str, charset: str) -> None:
+    """
+    Create a MySQL database if it doesn't already exist.
+
+    Executes CREATE DATABASE IF NOT EXISTS with the specified character set.
+    Safe to call multiple times - will not error if database already exists.
+
+    :param server_url: SQLAlchemy URL for MySQL server connection (without database selected)
+    :type server_url: sqlalchemy.engine.URL
+    :param db_name: Name of the database to create
+    :type db_name: str
+    :param charset: MySQL character set (e.g., 'latin1', 'utf8mb4')
+    :type charset: str
+    :return: None
+    :rtype: None
+    :raises ValueError: If db_name or charset contains invalid characters
+    """
+    # Validate database name to prevent SQL injection - only allow safe identifier characters
+    if not re.match(r'^[a-zA-Z0-9_$]+$', db_name):
+        raise ValueError(f"Invalid database name: {db_name}. Only alphanumeric, underscore, and dollar sign characters are allowed.")
+
+    # Validate charset name to prevent SQL injection - only allow safe characters
+    if not re.match(r'^[a-zA-Z0-9_]+$', charset):
+        raise ValueError(f"Invalid charset name: {charset}. Only alphanumeric and underscore characters are allowed.")
+
+    server_engine = create_engine(server_url)
+    with server_engine.begin() as conn:
+        # Safe to use f-string here since we've validated the inputs above
+        conn.execute(text(f"CREATE DATABASE IF NOT EXISTS `{db_name}` DEFAULT CHARACTER SET {charset}"))
+
+
+def ensure_schema(db_url: URL, spec, db_name: str) -> Dict[str, object]:
+    """
+    Create database tables and insert seed data if the table is empty.
+
+    Uses SQLAlchemy's metadata.create_all() to generate CREATE TABLE statements
+    from the schema specification. If seed_rows are defined in the spec and the
+    table is empty, inserts the seed data.
+
+    :param db_url: SQLAlchemy URL for the specific database connection
+    :type db_url: sqlalchemy.engine.URL
+    :param spec: Database schema specification from SIMPLE_EFP_DATABASE_SCHEMAS
+    :type spec: Dict[str, Any]
+    :param db_name: Name of the database (used for table naming)
+    :type db_name: str
+    :return: Dictionary with 'table' (table name) and 'seeded_rows' (count of inserted rows)
+    :rtype: Dict[str, object]
+    """
+    metadata = MetaData()
+    table = _build_table(metadata, spec, db_name)
+    engine = create_engine(db_url)
+    metadata.create_all(engine, checkfirst=True)
+
+    seed_rows = spec.get("seed_rows") or []
+    inserted = 0
+    if seed_rows:
+        with engine.connect() as conn:
+            count = conn.execute(text(f"SELECT COUNT(1) FROM {table.name}")).scalar() or 0
+            if count == 0:
+                with engine.begin() as write_conn:
+                    write_conn.execute(table.insert(), seed_rows)
+                    inserted = len(seed_rows)
+
+    return {"table": table.name, "seeded_rows": inserted}
+
+
+def bootstrap_simple_efp_databases(
+    *,
+    host: str,
+    port: int,
+    user: str,
+    password: str,
+    databases: Iterable[str] | None = None,
+) -> List[Dict[str, object]]:
+    """
+    Bootstrap simple eFP databases in MySQL from schema definitions.
+
+    This is the main entry point for creating eFP databases. For each database:
+    1. Creates the database if it doesn't exist
+    2. Creates the sample_data table with schema from SIMPLE_EFP_DATABASE_SCHEMAS
+    3. Inserts seed rows if the table is empty and seed_rows are defined
+
+    Used by:
+    - scripts/bootstrap_simple_efp_dbs.py (CLI tool)
+    - config/init.sh (Docker/CI initialization)
+    - api/resources/efp_proxy.py (HTTP bootstrap endpoint)
+
+    :param host: MySQL server hostname (e.g., 'localhost', 'BAR_mysqldb' for Docker)
+    :type host: str
+    :param port: MySQL server port number (typically 3306)
+    :type port: int
+    :param user: MySQL username with CREATE DATABASE privileges
+    :type user: str
+    :param password: MySQL password for authentication
+    :type password: str
+    :param databases: Optional list of specific databases to bootstrap; if None, bootstraps all databases in SIMPLE_EFP_DATABASE_SCHEMAS
+    :type databases: Iterable[str] or None
+    :return: List of result dictionaries, each containing 'database' (name), 'table' (table name), and 'seeded_rows' (count)
+    :rtype: List[Dict[str, object]]
+    :raises ValueError: If a requested database is not defined in SIMPLE_EFP_DATABASE_SCHEMAS
+
+    Example::
+
+        results = bootstrap_simple_efp_databases(
+            host='localhost',
+            port=3306,
+            user='root',
+            password='password',
+            databases=['cannabis', 'dna_damage']
+        )
+        # Returns: [
+        #     {'database': 'cannabis', 'table': 'sample_data', 'seeded_rows': 1},
+        #     {'database': 'dna_damage', 'table': 'sample_data', 'seeded_rows': 1}
+        # ]
+    """
+
+    results: List[Dict[str, object]] = []
+    target_dbs = list(databases) if databases is not None else list(SIMPLE_EFP_DATABASE_SCHEMAS.keys())
+
+    server_url = _build_url(host, port, user, password, database=None)
+    for db_name in target_dbs:
+        if db_name not in SIMPLE_EFP_DATABASE_SCHEMAS:
+            raise ValueError(f"Unknown simple eFP database: {db_name}")
+
+        spec = SIMPLE_EFP_DATABASE_SCHEMAS[db_name]
+        charset = spec.get("charset", "utf8mb4")
+        ensure_database(server_url, db_name, charset)
+        db_url = _build_url(host, port, user, password, database=db_name)
+        schema_result = ensure_schema(db_url, spec, db_name)
+        results.append(
+            {
+                "database": db_name,
+                "table": schema_result["table"],
+                "seeded_rows": schema_result["seeded_rows"],
+            }
+        )
+
+    return results
+
+
+__all__ = ["bootstrap_simple_efp_databases"]
diff --git a/scripts/bootstrap_simple_efp_dbs.py b/scripts/bootstrap_simple_efp_dbs.py
@@ -0,0 +1,98 @@
+#!/usr/bin/env python3
+"""
+bootstrap the simple efp databases from the shared schema definitions
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import sys
+from pathlib import Path
+
+from sqlalchemy.exc import SQLAlchemyError
+
+# ensure the repository root is importable when this script is executed standalone
+ROOT_DIR = Path(__file__).resolve().parents[1]
+if str(ROOT_DIR) not in sys.path:
+    sys.path.insert(0, str(ROOT_DIR))
+
+from api.services.efp_bootstrap import bootstrap_simple_efp_databases  # noqa: E402
+
+
+def _default_host() -> str:
+    """
+    Determine the default MySQL hostname from environment variables.
+
+    Priority order:
+    1. DB_HOST environment variable (explicit override)
+    2. MYSQL_HOST environment variable (alternative name)
+    3. 'localhost' (default for local dev and GitHub Actions CI)
+
+    Docker deployments should set DB_HOST=BAR_mysqldb explicitly.
+
+    :return: MySQL hostname to use for connections
+    :rtype: str
+    """
+    if os.environ.get("DB_HOST"):
+        return os.environ["DB_HOST"]
+    if os.environ.get("MYSQL_HOST"):
+        return os.environ["MYSQL_HOST"]
+    # GitHub Actions CI uses localhost; Docker Compose uses bar_mysqldb
+    # CI env var is set in GitHub Actions, but not DB_HOST
+    # In Docker, DB_HOST should be explicitly set to BAR_mysqldb
+    return "localhost"
+
+
+def parse_args() -> argparse.Namespace:
+    """
+    Parse command-line arguments for the bootstrap script.
+
+    :return: Parsed arguments containing host, port, user, password, and optional database list
+    :rtype: argparse.Namespace
+    """
+    parser = argparse.ArgumentParser(description="Create simple eFP MySQL databases from in-memory schemas.")
+    parser.add_argument("--host", default=_default_host(), help="MySQL hostname (default: %(default)s)")
+    parser.add_argument("--port", type=int, default=int(os.environ.get("DB_PORT", 3306)), help="MySQL port")
+    parser.add_argument("--user", default=os.environ.get("DB_USER", "root"), help="MySQL user")
+    parser.add_argument("--password", default=os.environ.get("DB_PASS", "root"), help="MySQL password")
+    parser.add_argument(
+        "--databases",
+        nargs="*",
+        help="Optional list of databases to bootstrap (defaults to every simple schema).",
+    )
+    return parser.parse_args()
+
+
+def main():
+    """
+    Main entry point for the bootstrap CLI script.
+
+    Parses command-line arguments and calls bootstrap_simple_efp_databases()
+    to create MySQL databases. Prints success messages for each database created.
+
+    Output format: [ok] ensured database_name.table_name (seeded N rows)
+
+    :raises SQLAlchemyError: If database creation or connection fails
+    """
+    args = parse_args()
+    results = bootstrap_simple_efp_databases(
+        host=args.host,
+        port=args.port,
+        user=args.user,
+        password=args.password,
+        databases=args.databases,
+    )
+    for entry in results:
+        seeded = entry["seeded_rows"]
+        seed_msg = f"seeded {seeded} rows" if seeded else "no seed rows inserted"
+        print(f"[ok] ensured {entry['database']}.{entry['table']} ({seed_msg})")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except SQLAlchemyError as exc:
+        # match init.sh style output to keep ci logs readable
+        print(f"failed to initialize simple efp databases: {exc}")
+        raise
