Spaces:
Running
Running
File size: 2,205 Bytes
9b788cc | 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 | import os
import sys
from dotenv import load_dotenv
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.core.config import configs
from app.core.database import Database
from app.core.security import get_password_hash
from app.model.user import Role, User
from app.util.hash import get_rand_hash
def _required_env(name: str) -> str:
value = os.getenv(name)
if not value:
raise RuntimeError(f"{name} must be set")
return value
def seed_admin() -> None:
load_dotenv()
email = _required_env("ADMIN_EMAIL")
password = _required_env("ADMIN_PASSWORD")
name = os.getenv("ADMIN_NAME", "System Admin")
identity_number = os.getenv("ADMIN_IDENTITY_NUMBER", "ADMIN-0001")
update_password = os.getenv("ADMIN_UPDATE_PASSWORD", "false").lower() in {"1", "true", "yes"}
db = Database(configs.DATABASE_URI)
with db.session() as session:
user = session.query(User).filter(User.email == email).first()
if user:
user.role = Role.admin
user.is_superuser = True
user.is_active = True
user.name = user.name or name
if update_password:
user.password = get_password_hash(password)
session.add(user)
session.commit()
print(f"Admin user already exists and was updated: {email}")
return
existing_identity = (
session.query(User)
.filter(User.identity_number == identity_number)
.first()
)
if existing_identity:
raise RuntimeError(
"ADMIN_IDENTITY_NUMBER is already used by another user. "
"Set a different ADMIN_IDENTITY_NUMBER."
)
admin = User(
email=email,
password=get_password_hash(password),
user_token=get_rand_hash(),
role=Role.admin,
identity_number=identity_number,
name=name,
is_active=True,
is_superuser=True,
)
session.add(admin)
session.commit()
print(f"Admin user created: {email}")
if __name__ == "__main__":
seed_admin()
|