3e90deba20
Changes:
- Removed server_default with CURRENT_TIMESTAMP from User model (SQLite doesn't support non-constant defaults in ALTER TABLE)
- Updated migrations 002 and 003 to add datetime columns without server defaults
- Added manual timestamp updates using SQLite's datetime('now') function in migrations
- Modified User model to handle timestamps in Python code instead of database defaults
- Updated profile routes to manually set updated_at timestamps on profile changes
- Enhanced User model __init__ to set default timestamps for new users
This resolves the 'Cannot add a column with non-constant default' SQLite error
while maintaining proper timestamp functionality.
108 lines
4.3 KiB
Python
108 lines
4.3 KiB
Python
"""Sync database with current model state
|
|
|
|
Revision ID: 003
|
|
Revises: 002
|
|
Create Date: 2024-01-01 00:00:02.000000
|
|
|
|
"""
|
|
from typing import Sequence, Union
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
from sqlalchemy.engine.reflection import Inspector
|
|
|
|
# revision identifiers, used by Alembic.
|
|
revision: str = '003'
|
|
down_revision: Union[str, None] = '002'
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
|
|
def table_exists(table_name):
|
|
"""Check if a table exists"""
|
|
bind = op.get_bind()
|
|
inspector = Inspector.from_engine(bind)
|
|
return table_name in inspector.get_table_names()
|
|
|
|
|
|
def column_exists(table_name, column_name):
|
|
"""Check if a column exists in a table"""
|
|
if not table_exists(table_name):
|
|
return False
|
|
bind = op.get_bind()
|
|
inspector = Inspector.from_engine(bind)
|
|
columns = [c['name'] for c in inspector.get_columns(table_name)]
|
|
return column_name in columns
|
|
|
|
|
|
def upgrade() -> None:
|
|
"""
|
|
This migration ensures the database is in sync with the current model state.
|
|
It safely handles the case where tables/columns may already exist.
|
|
"""
|
|
|
|
# Check if users table exists, if not create it with all fields
|
|
if not table_exists('users'):
|
|
op.create_table('users',
|
|
sa.Column('id', sa.Integer(), nullable=False),
|
|
sa.Column('email', sa.String(), nullable=False),
|
|
sa.Column('password_hash', sa.String(), nullable=False),
|
|
sa.Column('first_name', sa.String(), nullable=True),
|
|
sa.Column('last_name', sa.String(), nullable=True),
|
|
sa.Column('phone', sa.String(), nullable=True),
|
|
sa.Column('bio', sa.String(), nullable=True),
|
|
sa.Column('preferred_language', sa.String(), nullable=True),
|
|
sa.Column('timezone', sa.String(), nullable=True),
|
|
sa.Column('created_at', sa.DateTime(timezone=True), nullable=True), # Remove server_default for SQLite
|
|
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True), # Remove server_default for SQLite
|
|
sa.PrimaryKeyConstraint('id')
|
|
)
|
|
op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True)
|
|
op.create_index(op.f('ix_users_id'), 'users', ['id'], unique=False)
|
|
|
|
# Set default timestamps for new table
|
|
op.execute("UPDATE users SET created_at = datetime('now') WHERE created_at IS NULL")
|
|
op.execute("UPDATE users SET updated_at = datetime('now') WHERE updated_at IS NULL")
|
|
else:
|
|
# Table exists, check for missing profile columns and add them
|
|
profile_columns = [
|
|
('first_name', sa.Column('first_name', sa.String(), nullable=True)),
|
|
('last_name', sa.Column('last_name', sa.String(), nullable=True)),
|
|
('phone', sa.Column('phone', sa.String(), nullable=True)),
|
|
('bio', sa.Column('bio', sa.String(), nullable=True)),
|
|
('preferred_language', sa.Column('preferred_language', sa.String(), nullable=True)),
|
|
('timezone', sa.Column('timezone', sa.String(), nullable=True)),
|
|
('updated_at', sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True)) # Remove server_default for SQLite
|
|
]
|
|
|
|
for column_name, column_def in profile_columns:
|
|
if not column_exists('users', column_name):
|
|
op.add_column('users', column_def)
|
|
|
|
# Set default values for profile fields
|
|
if column_exists('users', 'preferred_language'):
|
|
op.execute("UPDATE users SET preferred_language = 'en' WHERE preferred_language IS NULL")
|
|
if column_exists('users', 'timezone'):
|
|
op.execute("UPDATE users SET timezone = 'UTC' WHERE timezone IS NULL")
|
|
if column_exists('users', 'updated_at'):
|
|
op.execute("UPDATE users SET updated_at = datetime('now') WHERE updated_at IS NULL")
|
|
|
|
|
|
def downgrade() -> None:
|
|
"""
|
|
Downgrade removes the profile fields added in this migration
|
|
"""
|
|
if table_exists('users'):
|
|
profile_columns = [
|
|
'updated_at',
|
|
'timezone',
|
|
'preferred_language',
|
|
'bio',
|
|
'phone',
|
|
'last_name',
|
|
'first_name'
|
|
]
|
|
|
|
for column_name in profile_columns:
|
|
if column_exists('users', column_name):
|
|
op.drop_column('users', column_name) |