from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# Configure the database URL (SQLite for simplicity)
DATABASE_URL = "sqlite:///test.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
from app import Base
from sqlalchemy import Column, Integer, String
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
email = Column(String, unique=True, index=True)
alembic init alembic
sqlalchemy.url = sqlite:///test.db
from app import Base
# Add this to the run_migrations_online function
from app import engine
target_metadata = Base.metadata
import os
from alembic.config import Config
from alembic import command
def init():
"""Initialize Alembic migration environment."""
print("Alembic has already been initialized. Run 'alembic init' directly if needed.")
def migrate():
"""Generate a new migration."""
alembic_cfg = Config("alembic.ini")
command.revision(alembic_cfg, autogenerate=True, message="Initial migration")
def upgrade():
"""Apply migrations."""
alembic_cfg = Config("alembic.ini")
command.upgrade(alembic_cfg, "head")
def downgrade():
"""Revert migrations."""
alembic_cfg = Config("alembic.ini")
command.downgrade(alembic_cfg, "-1")
if __name__ == "__main__":
import sys
commands = {
"db init": init,
"db migrate": migrate,
"db upgrade": upgrade,
"db downgrade": downgrade,
}
if len(sys.argv) < 2:
print("Usage: python manage.py <command>")
elif sys.argv[1] in commands:
commands[sys.argv[1]]()
else:
print(f"Unknown command: {sys.argv[1]}")