Spaces:
Sleeping
Sleeping
File size: 5,182 Bytes
e327f0d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 | """
backend/scripts/db_init.py
One-shot DB init: Alembic'i programatik calistirir ve istege bagli admin user yaratir.
Calistirma (services/backend altinda):
python -m scripts.db_init
# veya
python scripts/db_init.py
Env'ler:
DATABASE_URL / DATABASE_URL_ASYNC — zorunlu
ADMIN_EMAIL — opsiyonel (varsa ADMIN_PASSWORD da gerekli)
ADMIN_PASSWORD — opsiyonel
Sifre hashing:
`security.py` Backend Architect tarafindan yazildiginda `hash_password`
fonksiyonu oradan import edilir. Su an direkt passlib[bcrypt] kullanilir
(requirements.txt'te zaten mevcut).
"""
from __future__ import annotations
import asyncio
import logging
import os
import sys
from pathlib import Path
# services/backend dizinini path'e ekle
_BACKEND_DIR = Path(__file__).resolve().parent.parent
if str(_BACKEND_DIR) not in sys.path:
sys.path.insert(0, str(_BACKEND_DIR))
from sqlalchemy import select # noqa: E402
from database import AsyncSessionLocal, engine # noqa: E402
from db_models import User, UserRole # noqa: E402
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)-5s [%(name)s] %(message)s",
)
log = logging.getLogger("db_init")
# ---------------- Password hashing ----------------
def _hash_password(plain: str) -> str:
"""Bcrypt hash.
Once `security.py`'den import etmeyi dene (Backend Architect tarafindan
yazilacak kanonik fonksiyon). Yoksa passlib ile direkt bcrypt.
"""
try:
from security import hash_password # type: ignore
return hash_password(plain)
except Exception:
log.debug("security.hash_password yok — passlib bcrypt fallback")
try:
from passlib.context import CryptContext
ctx = CryptContext(schemes=["bcrypt"], deprecated="auto")
return ctx.hash(plain)
except Exception as exc: # pragma: no cover
raise RuntimeError(
"Sifre hashleyemedim. Backend Architect 'security.py' icine "
"`hash_password(plain: str) -> str` bcrypt fonksiyonu yazmali "
"veya passlib[bcrypt] kurulu olmali."
) from exc
# ---------------- Alembic upgrade ----------------
def _run_migrations() -> None:
"""`alembic upgrade head` esdegerini programatik calistir."""
from alembic import command
from alembic.config import Config
ini_path = _BACKEND_DIR / "alembic.ini"
if not ini_path.exists():
raise FileNotFoundError(f"alembic.ini bulunamadi: {ini_path}")
cfg = Config(str(ini_path))
cfg.set_main_option("script_location", str(_BACKEND_DIR / "migrations"))
log.info("Alembic upgrade head calisiyor...")
command.upgrade(cfg, "head")
log.info("Migration tamam.")
# ---------------- Admin seed ----------------
async def _ensure_admin() -> None:
admin_email = os.getenv("ADMIN_EMAIL")
admin_password = os.getenv("ADMIN_PASSWORD")
if not admin_email:
log.info("ADMIN_EMAIL set degil — admin user yaratilmadi (atlandi).")
return
if not admin_password:
log.warning(
"ADMIN_EMAIL var ama ADMIN_PASSWORD yok. Admin yaratilmadi. "
"Iki env'i de doldur."
)
return
admin_email_norm = admin_email.strip().lower()
async with AsyncSessionLocal() as session:
existing = await session.scalar(
select(User).where(User.email == admin_email_norm)
)
if existing is not None:
log.info("Admin user zaten var: %s (id=%s)", admin_email_norm, existing.id)
return
user = User(
email=admin_email_norm,
password_hash=_hash_password(admin_password),
full_name=os.getenv("ADMIN_FULL_NAME", "Admin"),
role=UserRole.ADMIN,
is_active=True,
)
session.add(user)
await session.commit()
await session.refresh(user)
log.info("Admin user olusturuldu: %s (id=%s)", admin_email_norm, user.id)
# ---------------- Healthcheck ----------------
async def _ping() -> None:
from sqlalchemy import text
async with engine.connect() as conn:
result = await conn.execute(text("SELECT 1"))
assert result.scalar() == 1
log.info("DB ping OK.")
# ---------------- Main ----------------
async def main() -> None:
log.info("DB init basliyor (DATABASE_URL=%s)", _mask_url())
await _ping()
_run_migrations()
await _ensure_admin()
await engine.dispose()
log.info("DB init tamam.")
def _mask_url() -> str:
url = os.getenv("DATABASE_URL_ASYNC") or os.getenv("DATABASE_URL") or "<unset>"
# parola maskele
if "@" in url and "://" in url:
scheme, rest = url.split("://", 1)
if "@" in rest:
creds, host = rest.split("@", 1)
if ":" in creds:
user, _ = creds.split(":", 1)
return f"{scheme}://{user}:***@{host}"
return url
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
sys.exit(130)
except Exception as exc:
log.exception("DB init basarisiz: %s", exc)
sys.exit(1)
|