GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-08-28 03:27:31 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix db

Browse Source
This commit is contained in:
jxxghp
2025-08-19 12:35:52 +08:00
parent be08d6ebb5
commit cf28d93be6
2 changed files with 107 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
database/versions/5b3355c964bb_2_2_0.py
+106
View File
@@ -0,0 +1,106 @@
"""2.2.0
Revision ID: 5b3355c964bb
Revises: d58298a0879f
Create Date: 2025-08-19 12:27:08.451371
"""
import sqlalchemy as sa
from alembic import op
from app.core.config import settings
# revision identifiers, used by Alembic.
revision = '5b3355c964bb'
down_revision = 'd58298a0879f'
branch_labels = None
depends_on = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
if settings.DB_TYPE.lower() == "postgresql":
# 将SQLite的Sequence转换为PostgreSQL的Identity
fix_postgresql_sequences()
# ### end Alembic commands ###
def fix_postgresql_sequences():
"""
修复PostgreSQL数据库中的序列问题
将SQLite迁移过来的Sequence转换为PostgreSQL的Identity
"""
connection = op.get_bind()
# 获取所有表名
result = connection.execute(sa.text("""
SELECT table_name
FROM information_schema.tables
WHERE table_schema = 'public'
AND table_type = 'BASE TABLE'
"""))
tables = [row[0] for row in result.fetchall()]
print(f"发现 {len(tables)} 个表需要检查序列")
for table_name in tables:
fix_table_sequence(connection, table_name)
def fix_table_sequence(connection, table_name):
"""
修复单个表的序列
"""
try:
# 检查表是否有id列
result = connection.execute(sa.text(f"""
SELECT column_name, data_type, is_nullable, column_default
FROM information_schema.columns
WHERE table_name = '{table_name}'
AND column_name = 'id'
"""))
id_column = result.fetchone()
if not id_column:
print(f"表 {table_name} 没有id列，跳过")
return
_, _, _, column_default = id_column
# 检查是否已经是Identity类型
if column_default and 'GENERATED BY DEFAULT AS IDENTITY' in column_default:
print(f"表 {table_name} 的id列已经是Identity类型，跳过")
return
# 检查是否有序列
print(f"表 {table_name} 存在序列，需要修复")
convert_to_identity(connection, table_name)
except Exception as e:
print(f"修复表 {table_name} 序列时出错: {e}")
def convert_to_identity(connection, table_name):
"""
将序列转换为Identity，保持原有约束不变
"""
try:
# 获取当前序列的最大值
result = connection.execute(sa.text(f"""
SELECT COALESCE(MAX(id), 0) + 1 as next_value
FROM {table_name}
"""))
next_value = result.fetchone()[0]
# 直接修改列属性，添加Identity，保持其他约束不变
# 这种方式不会删除主键约束和索引
connection.execute(sa.text(f"""
ALTER TABLE {table_name}
ALTER COLUMN id ADD GENERATED BY DEFAULT AS IDENTITY (START WITH {next_value})
"""))
print(f"表 {table_name} 序列已转换为Identity，起始值为 {next_value}")
except Exception as e:
print(f"转换表 {table_name} 序列时出错: {e}")
raise
database/versions/d58298a0879f_2_1_9.py
+1 -105
View File
@@ -5,12 +5,6 @@ Revises: 4666ce24a443
Create Date: 2025-08-19 11:56:39.652032 Create Date: 2025-08-19 11:56:39.652032
""" """
import contextlib
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import sqlite
from app.core.config import settings
# revision identifiers, used by Alembic. # revision identifiers, used by Alembic.
revision = 'd58298a0879f' revision = 'd58298a0879f'
@@ -20,106 +14,8 @@ depends_on = None
def upgrade() -> None: def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ### pass
if settings.DB_TYPE.lower() == "postgresql":
# PostgreSQL序列修复：将SQLite的Sequence转换为PostgreSQL的Identity
fix_postgresql_sequences()
# ### end Alembic commands ###
def downgrade() -> None: def downgrade() -> None:
pass pass
def fix_postgresql_sequences():
"""
修复PostgreSQL数据库中的序列问题
将SQLite迁移过来的Sequence转换为PostgreSQL的Identity
"""
connection = op.get_bind()
# 获取所有表名
result = connection.execute(sa.text("""
SELECT table_name
FROM information_schema.tables
WHERE table_schema = 'public'
AND table_type = 'BASE TABLE'
"""))
tables = [row[0] for row in result.fetchall()]
print(f"发现 {len(tables)} 个表需要检查序列")
for table_name in tables:
fix_table_sequence(connection, table_name)
def fix_table_sequence(connection, table_name):
"""
修复单个表的序列
"""
try:
# 检查表是否有id列
result = connection.execute(sa.text(f"""
SELECT column_name, data_type, is_nullable, column_default
FROM information_schema.columns
WHERE table_name = '{table_name}'
AND column_name = 'id'
"""))
id_column = result.fetchone()
if not id_column:
print(f"表 {table_name} 没有id列，跳过")
return
column_name, data_type, is_nullable, column_default = id_column
# 检查是否已经是Identity类型
if column_default and 'GENERATED BY DEFAULT AS IDENTITY' in column_default:
print(f"表 {table_name} 的id列已经是Identity类型，跳过")
return
# 检查是否有序列
result = connection.execute(sa.text(f"""
SELECT sequence_name
FROM information_schema.sequences
WHERE sequence_name LIKE '{table_name}_id_seq'
"""))
sequence_exists = result.fetchone()
if sequence_exists:
print(f"表 {table_name} 存在序列，需要修复")
convert_to_identity(connection, table_name)
except Exception as e:
print(f"修复表 {table_name} 序列时出错: {e}")
def convert_to_identity(connection, table_name):
"""
将序列转换为Identity
"""
try:
# 获取当前序列的最大值
result = connection.execute(sa.text(f"""
SELECT COALESCE(MAX(id), 0) + 1 as next_value
FROM {table_name}
"""))
next_value = result.fetchone()[0]
# 删除旧的序列
connection.execute(sa.text(f"DROP SEQUENCE IF EXISTS {table_name}_id_seq"))
# 修改列定义，添加Identity
connection.execute(sa.text(f"""
ALTER TABLE {table_name}
ALTER COLUMN id SET DATA TYPE INTEGER,
ALTER COLUMN id SET DEFAULT NULL,
ALTER COLUMN id ADD GENERATED BY DEFAULT AS IDENTITY (START WITH {next_value})
"""))
print(f"表 {table_name} 序列已转换为Identity，起始值为 {next_value}")
except Exception as e:
print(f"转换表 {table_name} 序列时出错: {e}")
raise