Cheng Zhou
917ab7ccf1
Add early termination visit workflow with ordering, non-applicable visit handling, visit window display, and medication adherence support. Extend monitoring visit issue template fields, site scoping, setup draft project info handling, login security UI, attachment behavior, and related tests/migrations.
60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
import asyncio
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import inspect
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from app.core.config import PROTECTED_ADMIN_EMAIL # noqa: E402
|
|
from app.crud.user import ensure_admin_exists # noqa: E402
|
|
from app.db.base import Base # noqa: E402
|
|
from app.db.session import SessionLocal, engine # noqa: E402
|
|
|
|
|
|
def run_alembic_command(*args: str) -> None:
|
|
subprocess.run(
|
|
[sys.executable, "-m", "alembic", *args],
|
|
check=True,
|
|
cwd=PROJECT_ROOT,
|
|
)
|
|
|
|
|
|
def _list_user_tables(sync_conn) -> list[str]:
|
|
inspector = inspect(sync_conn)
|
|
return [table for table in inspector.get_table_names(schema="public") if table != "alembic_version"]
|
|
|
|
|
|
async def initialize_schema() -> None:
|
|
async with engine.begin() as conn:
|
|
user_tables = await conn.run_sync(_list_user_tables)
|
|
if user_tables:
|
|
run_alembic_command("upgrade", "head")
|
|
return
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
run_alembic_command("stamp", "head")
|
|
|
|
|
|
async def seed_protected_admin() -> None:
|
|
async with SessionLocal() as session:
|
|
await ensure_admin_exists(session)
|
|
|
|
|
|
async def async_main() -> None:
|
|
print("Initializing database schema...")
|
|
await initialize_schema()
|
|
print(f"Ensuring protected admin exists: {PROTECTED_ADMIN_EMAIL}")
|
|
await seed_protected_admin()
|
|
print("Production initialization complete.")
|
|
|
|
|
|
def main() -> None:
|
|
asyncio.run(async_main())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|