cb5bfcf7b5
All checks were successful
Deploy to Production, build images and push to Gitea Registry / build_and_push (pull_request) Successful in 1m17s
57 lines
1.8 KiB
Python
57 lines
1.8 KiB
Python
"""
|
|
Async migrations handler for FastAPI application.
|
|
This file is separate from env.py to avoid Alembic context issues.
|
|
"""
|
|
import os
|
|
import sys
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
from sqlalchemy import pool
|
|
|
|
# Ensure the app directory is in the Python path
|
|
sys.path.insert(0, os.path.realpath(os.path.join(os.path.dirname(__file__), '..')))
|
|
|
|
from app.database import Base as DatabaseBase
|
|
from app.config import settings
|
|
|
|
async def run_migrations():
|
|
"""Run database migrations asynchronously."""
|
|
from alembic.runtime.migration import MigrationContext
|
|
from alembic.operations import Operations
|
|
|
|
# Create async engine
|
|
engine = create_async_engine(
|
|
settings.DATABASE_URL,
|
|
poolclass=pool.NullPool,
|
|
)
|
|
|
|
async with engine.connect() as connection:
|
|
# Get current database schema version
|
|
def do_get_current_rev():
|
|
migration_context = MigrationContext.configure(
|
|
connection,
|
|
opts={
|
|
'target_metadata': DatabaseBase.metadata,
|
|
'compare_type': True,
|
|
'compare_server_default': True
|
|
}
|
|
)
|
|
return migration_context.get_current_revision()
|
|
|
|
current_rev = await connection.run_sync(do_get_current_rev)
|
|
|
|
# Run migrations
|
|
def do_upgrade():
|
|
migration_context = MigrationContext.configure(
|
|
connection,
|
|
opts={
|
|
'target_metadata': DatabaseBase.metadata,
|
|
'compare_type': True,
|
|
'compare_server_default': True
|
|
}
|
|
)
|
|
with Operations.context(migration_context):
|
|
migration_context.run_migrations()
|
|
|
|
await connection.run_sync(do_upgrade)
|
|
|
|
await engine.dispose() |