feat: 扩展多个字符串列为 TEXT 类型并添加 tls-client 可选依赖

数据库迁移:
- 扩展 provider_api_keys.api_key 为 TEXT(OAuth tokens 可能很长)
- 扩展 ldap_configs 的 bind_dn、base_dn、user_search_filter 为 TEXT
- 扩展 oauth_providers.client_id 为 TEXT
- 添加 SQLite 兼容支持(batch 模式)
- 添加表/列存在性检查

依赖:
- 添加 tls-client 作为可选依赖 [tls]
This commit is contained in:
fawney19
2026-02-04 16:45:05 +08:00
parent c996078f30
commit 24c9105628
3 changed files with 104 additions and 25 deletions

View File

@@ -1,7 +1,7 @@
"""Add provider_type and expand api_key column to TEXT
"""Add provider_type and expand string columns to TEXT
- Add providers.provider_type (String(20), server_default="custom")
- Change provider_api_keys.api_key from VARCHAR(500) to TEXT (OAuth tokens can be long)
- Expand multiple VARCHAR columns to TEXT for long values (OAuth tokens, LDAP DN, URLs, etc.)
Revision ID: b5c6d7e8f9a0
Revises: c4e8f9a1b2c3
@@ -22,6 +22,16 @@ branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
# 需要扩展为 TEXT 的列(表名, 列名, 原始类型长度)
COLUMNS_TO_EXPAND = [
("provider_api_keys", "api_key", 500), # OAuth tokens can be very long
("ldap_configs", "bind_dn", 255), # LDAP DN can be deeply nested
("ldap_configs", "base_dn", 255), # LDAP DN can be deeply nested
("ldap_configs", "user_search_filter", 500), # Complex LDAP filters
("oauth_providers", "client_id", 255), # Some OAuth providers use JWT client_id
]
def column_exists(table_name: str, column_name: str) -> bool:
"""检查列是否已存在"""
bind = op.get_bind()
@@ -30,6 +40,72 @@ def column_exists(table_name: str, column_name: str) -> bool:
return column_name in columns
def table_exists(table_name: str) -> bool:
"""检查表是否存在"""
bind = op.get_bind()
inspector = inspect(bind)
return table_name in inspector.get_table_names()
def is_sqlite() -> bool:
"""检查是否为 SQLite 数据库"""
bind = op.get_bind()
return bind.dialect.name == "sqlite"
def expand_column_to_text(table_name: str, column_name: str, original_length: int) -> None:
"""将 VARCHAR 列扩展为 TEXT兼容 SQLite"""
if not table_exists(table_name):
return
if not column_exists(table_name, column_name):
return
if is_sqlite():
# SQLite 不支持直接 ALTER COLUMN需要用 batch 模式
with op.batch_alter_table(table_name) as batch_op:
batch_op.alter_column(
column_name,
type_=sa.Text(),
existing_type=sa.String(original_length),
)
else:
op.alter_column(
table_name,
column_name,
type_=sa.Text(),
existing_type=sa.String(original_length),
)
def shrink_column_to_varchar(
table_name: str, column_name: str, target_length: int, nullable: bool = False
) -> None:
"""将 TEXT 列缩小为 VARCHAR兼容 SQLite
WARNING: 如果数据超过 target_length 会失败
"""
if not table_exists(table_name):
return
if not column_exists(table_name, column_name):
return
if is_sqlite():
with op.batch_alter_table(table_name) as batch_op:
batch_op.alter_column(
column_name,
type_=sa.String(target_length),
existing_type=sa.Text(),
existing_nullable=nullable,
)
else:
op.alter_column(
table_name,
column_name,
type_=sa.String(target_length),
existing_type=sa.Text(),
existing_nullable=nullable,
)
def upgrade() -> None:
# Add providers.provider_type
if not column_exists("providers", "provider_type"):
@@ -38,26 +114,16 @@ def upgrade() -> None:
sa.Column("provider_type", sa.String(20), nullable=False, server_default="custom"),
)
# Expand provider_api_keys.api_key from VARCHAR(500) to TEXT
op.alter_column(
"provider_api_keys",
"api_key",
type_=sa.Text(),
existing_type=sa.String(500),
existing_nullable=False,
)
# Expand VARCHAR columns to TEXT
for table_name, column_name, original_length in COLUMNS_TO_EXPAND:
expand_column_to_text(table_name, column_name, original_length)
def downgrade() -> None:
# Revert provider_api_keys.api_key from TEXT to VARCHAR(500)
# WARNING: Downgrade may fail if any api_key values exceed 500 characters
op.alter_column(
"provider_api_keys",
"api_key",
type_=sa.String(500),
existing_type=sa.Text(),
existing_nullable=False,
)
# Shrink TEXT columns back to VARCHAR
# WARNING: Downgrade may fail if any values exceed original length
for table_name, column_name, original_length in reversed(COLUMNS_TO_EXPAND):
shrink_column_to_varchar(table_name, column_name, original_length)
# Drop providers.provider_type
if column_exists("providers", "provider_type"):