cardosofelipe/syndarix
1
0
Fork 0
forked from cardosofelipe/fast-next-template
Code Issues 18 Pull Requests Wiki Activity

refactor(migrations): replace hardcoded database URL with configurable environment variable and update command syntax to use consistent quoting style

Browse Source
This commit is contained in:
Felipe Cardoso
2026-01-06 17:19:28 +01:00
parent a0ec5fa2cc
commit 3f23bc3db3
1 changed files with 47 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
backend/migrate.py
View File

@@ -26,6 +26,7 @@ Usage:
# Inside Docker (without --local flag): # Inside Docker (without --local flag):
python migrate.py auto "Add new field" python migrate.py auto "Add new field"
""" """
import argparse import argparse
import os import os
import subprocess import subprocess
@@ -44,13 +45,14 @@ def setup_database_url(use_local: bool) -> str:
# Override DATABASE_URL to use localhost instead of Docker hostname # Override DATABASE_URL to use localhost instead of Docker hostname
local_url = os.environ.get( local_url = os.environ.get(
"LOCAL_DATABASE_URL", "LOCAL_DATABASE_URL",
"postgresql://postgres:postgres@localhost:5432/app" "postgresql://postgres:postgres@localhost:5432/syndarix",
) )
os.environ["DATABASE_URL"] = local_url os.environ["DATABASE_URL"] = local_url
return local_url return local_url
# Use the configured DATABASE_URL from environment/.env # Use the configured DATABASE_URL from environment/.env
from app.core.config import settings from app.core.config import settings
return settings.database_url return settings.database_url
@@ -61,6 +63,7 @@ def check_models():
try: try:
# Import all models through the models package # Import all models through the models package
from app.models import __all__ as all_models from app.models import __all__ as all_models
print(f"Found {len(all_models)} model(s):") print(f"Found {len(all_models)} model(s):")
for model in all_models: for model in all_models:
print(f" - {model}") print(f" - {model}")
@@ -110,7 +113,9 @@ def generate_migration(message, rev_id=None, auto_rev_id=True, offline=False):
# Look for the revision ID, which is typically 12 hex characters # Look for the revision ID, which is typically 12 hex characters
parts = line.split() parts = line.split()
for part in parts: for part in parts:
if len(part) >= 12 and all(c in "0123456789abcdef" for c in part[:12]): if len(part) >= 12 and all(
c in "0123456789abcdef" for c in part[:12]
):
revision = part[:12] revision = part[:12]
break break
except Exception as e: except Exception as e:
@@ -185,6 +190,7 @@ def check_database_connection():
db_url = os.environ.get("DATABASE_URL") db_url = os.environ.get("DATABASE_URL")
if not db_url: if not db_url:
from app.core.config import settings from app.core.config import settings
db_url = settings.database_url db_url = settings.database_url
engine = create_engine(db_url) engine = create_engine(db_url)
@@ -270,8 +276,8 @@ def generate_offline_migration(message, rev_id):
content = f'''"""{message} content = f'''"""{message}
Revision ID: {rev_id} Revision ID: {rev_id}
Revises: {down_revision or ''} Revises: {down_revision or ""}
Create Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')} Create Date: {datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")}
""" """
@@ -320,6 +326,7 @@ def reset_alembic_version():
db_url = os.environ.get("DATABASE_URL") db_url = os.environ.get("DATABASE_URL")
if not db_url: if not db_url:
from app.core.config import settings from app.core.config import settings
db_url = settings.database_url db_url = settings.database_url
try: try:
@@ -338,82 +345,80 @@ def reset_alembic_version():
def main(): def main():
"""Main function""" """Main function"""
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description='Database migration helper for Generative Models Arena' description="Database migration helper for Generative Models Arena"
) )
# Global options # Global options
parser.add_argument( parser.add_argument(
'--local', '-l', "--local",
action='store_true', "-l",
help='Use localhost instead of Docker hostname (for local development)' action="store_true",
help="Use localhost instead of Docker hostname (for local development)",
) )
subparsers = parser.add_subparsers(dest='command', help='Command to run') subparsers = parser.add_subparsers(dest="command", help="Command to run")
# Generate command # Generate command
generate_parser = subparsers.add_parser('generate', help='Generate a migration') generate_parser = subparsers.add_parser("generate", help="Generate a migration")
generate_parser.add_argument('message', help='Migration message') generate_parser.add_argument("message", help="Migration message")
generate_parser.add_argument( generate_parser.add_argument(
'--rev-id', "--rev-id", help="Custom revision ID (e.g., 0001, 0002 for sequential naming)"
help='Custom revision ID (e.g., 0001, 0002 for sequential naming)'
) )
generate_parser.add_argument( generate_parser.add_argument(
'--offline', "--offline",
action='store_true', action="store_true",
help='Generate empty migration template without database connection' help="Generate empty migration template without database connection",
) )
# Apply command # Apply command
apply_parser = subparsers.add_parser('apply', help='Apply migrations') apply_parser = subparsers.add_parser("apply", help="Apply migrations")
apply_parser.add_argument('--revision', help='Specific revision to apply to') apply_parser.add_argument("--revision", help="Specific revision to apply to")
# List command # List command
subparsers.add_parser('list', help='List migrations') subparsers.add_parser("list", help="List migrations")
# Current command # Current command
subparsers.add_parser('current', help='Show current revision') subparsers.add_parser("current", help="Show current revision")
# Check command # Check command
subparsers.add_parser('check', help='Check database connection and models') subparsers.add_parser("check", help="Check database connection and models")
# Next command (show next revision ID) # Next command (show next revision ID)
subparsers.add_parser('next', help='Show the next sequential revision ID') subparsers.add_parser("next", help="Show the next sequential revision ID")
# Reset command (clear alembic_version table) # Reset command (clear alembic_version table)
subparsers.add_parser( subparsers.add_parser(
'reset', "reset", help="Reset alembic_version table (use after deleting all migrations)"
help='Reset alembic_version table (use after deleting all migrations)'
) )
# Auto command (generate and apply) # Auto command (generate and apply)
auto_parser = subparsers.add_parser('auto', help='Generate and apply migration') auto_parser = subparsers.add_parser("auto", help="Generate and apply migration")
auto_parser.add_argument('message', help='Migration message') auto_parser.add_argument("message", help="Migration message")
auto_parser.add_argument( auto_parser.add_argument(
'--rev-id', "--rev-id", help="Custom revision ID (e.g., 0001, 0002 for sequential naming)"
help='Custom revision ID (e.g., 0001, 0002 for sequential naming)'
) )
auto_parser.add_argument( auto_parser.add_argument(
'--offline', "--offline",
action='store_true', action="store_true",
help='Generate empty migration template without database connection' help="Generate empty migration template without database connection",
) )
args = parser.parse_args() args = parser.parse_args()
# Commands that don't need database connection # Commands that don't need database connection
if args.command == 'next': if args.command == "next":
show_next_rev_id() show_next_rev_id()
return return
# Check if offline mode is requested # Check if offline mode is requested
offline = getattr(args, 'offline', False) offline = getattr(args, "offline", False)
# Offline generate doesn't need database or model check # Offline generate doesn't need database or model check
if args.command == 'generate' and offline: if args.command == "generate" and offline:
generate_migration(args.message, rev_id=args.rev_id, offline=True) generate_migration(args.message, rev_id=args.rev_id, offline=True)
return return
if args.command == 'auto' and offline: if args.command == "auto" and offline:
generate_migration(args.message, rev_id=args.rev_id, offline=True) generate_migration(args.message, rev_id=args.rev_id, offline=True)
print("\nOffline migration generated. Apply it later with:") print("\nOffline migration generated. Apply it later with:")
print(" python migrate.py --local apply") print(" python migrate.py --local apply")
@@ -423,27 +428,27 @@ def main():
db_url = setup_database_url(args.local) db_url = setup_database_url(args.local)
print(f"Using database URL: {db_url}") print(f"Using database URL: {db_url}")
if args.command == 'generate': if args.command == "generate":
check_models() check_models()
generate_migration(args.message, rev_id=args.rev_id) generate_migration(args.message, rev_id=args.rev_id)
elif args.command == 'apply': elif args.command == "apply":
apply_migration(args.revision) apply_migration(args.revision)
elif args.command == 'list': elif args.command == "list":
list_migrations() list_migrations()
elif args.command == 'current': elif args.command == "current":
show_current() show_current()
elif args.command == 'check': elif args.command == "check":
check_database_connection() check_database_connection()
check_models() check_models()
elif args.command == 'reset': elif args.command == "reset":
reset_alembic_version() reset_alembic_version()
elif args.command == 'auto': elif args.command == "auto":
check_models() check_models()
revision = generate_migration(args.message, rev_id=args.rev_id) revision = generate_migration(args.message, rev_id=args.rev_id)
if revision: if revision: