|
7 | 7 | from __future__ import annotations |
8 | 8 |
|
9 | 9 | import re |
| 10 | +import hashlib |
10 | 11 | from typing import Dict, Iterable, List |
11 | 12 |
|
12 | 13 | from sqlalchemy import Column, Index, MetaData, Table, create_engine, text |
@@ -73,10 +74,30 @@ def _build_table(metadata: MetaData, spec, db_name: str) -> Table: |
73 | 74 | table = Table(spec["table_name"], metadata, *columns, mysql_charset=spec.get("charset")) |
74 | 75 | index_cols = spec.get("index") or [] |
75 | 76 | if index_cols: |
76 | | - Index(f"ix_{db_name}_{'_'.join(index_cols)}", *[table.c[col] for col in index_cols]) |
| 77 | + index_name = _make_index_name(db_name, index_cols) |
| 78 | + Index(index_name, *[table.c[col] for col in index_cols]) |
77 | 79 | return table |
78 | 80 |
|
79 | 81 |
|
| 82 | +def _make_index_name(db_name: str, index_cols: Iterable[str], max_len: int = 64) -> str: |
| 83 | + """ |
| 84 | + Create a MySQL-safe index name capped at 64 characters. |
| 85 | +
|
| 86 | + If the generated name is too long, fall back to a truncated db_name with a stable hash |
| 87 | + to keep names deterministic and avoid collisions. |
| 88 | + """ |
| 89 | + base = f"ix_{db_name}_{'_'.join(index_cols)}" |
| 90 | + if len(base) <= max_len: |
| 91 | + return base |
| 92 | + |
| 93 | + digest = hashlib.sha1(base.encode("utf-8")).hexdigest()[:8] |
| 94 | + reserved = len("ix_") + 1 + len(digest) |
| 95 | + db_len = max_len - reserved |
| 96 | + if db_len <= 0: |
| 97 | + return f"ix_{digest}" |
| 98 | + return f"ix_{db_name[:db_len]}_{digest}" |
| 99 | + |
| 100 | + |
80 | 101 | def _build_url(host: str, port: int, user: str, password: str, database: str | None = None) -> URL: |
81 | 102 | """ |
82 | 103 | Build a SQLAlchemy database URL for MySQL connections. |
|
0 commit comments