88 lines
3.6 KiB
Python
88 lines
3.6 KiB
Python
from sqlalchemy.exc import IntegrityError
|
|
from datetime import datetime as dt, timezone as tz
|
|
from flask_security import hash_password
|
|
from uuid import uuid4
|
|
|
|
from eveai_app import create_app
|
|
from common.extensions import db, security
|
|
from common.models.user import User, Tenant, Role, RolesUsers
|
|
from common.utils.database import Database
|
|
|
|
|
|
def initialize_data():
|
|
app = create_app()
|
|
with app.app_context():
|
|
# Define Initial Tenant
|
|
if Tenant.query.first() is None:
|
|
print("No tenants found. Creating initial tenant...")
|
|
timestamp = dt.now(tz=tz.utc)
|
|
tenant = Tenant(name="Jedi",
|
|
website="https://askeveai.com",
|
|
created_at=timestamp,
|
|
updated_at=timestamp)
|
|
db.session.add(tenant)
|
|
db.session.commit()
|
|
else:
|
|
print("Tenants already exist. Skipping tenant creation.")
|
|
|
|
Database(tenant.id).create_tenant_schema()
|
|
|
|
# Define Roles
|
|
super_user_role = Role.query.filter_by(name="Super User").first()
|
|
if super_user_role is None:
|
|
super_user_role = Role(name="Super User", description="Users allowed to perform all functions")
|
|
db.session.add(super_user_role)
|
|
db.session.commit()
|
|
tenant_admin_role = Role.query.filter_by(name="Tenant Admin").first()
|
|
if tenant_admin_role is None:
|
|
tenant_admin_role = Role(name="Tenant Admin", description="Users allowed to manage tenants")
|
|
db.session.add(tenant_admin_role)
|
|
db.session.commit()
|
|
# tenant_tester_role = Role.query.filter_by(name="Tenant Tester").first()
|
|
# if tenant_tester_role is None:
|
|
# tenant_test_role = Role(name="Tenant Tester", description="Users allowed to test tenants")
|
|
# db.session.add(tenant_test_role)
|
|
# db.session.commit()
|
|
|
|
# Check if any users exist
|
|
if User.query.first() is None:
|
|
print("No users found. Creating initial user...")
|
|
# Ensure tenant exists before creating the user
|
|
tenant = Tenant.query.filter_by(name="Jedi").first()
|
|
if tenant:
|
|
user = User(
|
|
user_name="yoda",
|
|
email="yoda@flow-it.net",
|
|
password=hash_password("Dagobah"),
|
|
first_name="Yoda",
|
|
last_name="Skywalker",
|
|
tenant_id=tenant.id,
|
|
created_at=dt.now(tz=tz.utc),
|
|
updated_at=dt.now(tz=tz.utc),
|
|
fs_uniquifier=str(uuid4()),
|
|
active=True,
|
|
confirmed_at=dt.now(tz=tz.utc)
|
|
)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
# security.datastore.set_uniquifier()
|
|
print("Initial user created.")
|
|
# Assign SuperUser role to the new user
|
|
user_role = RolesUsers(user_id=user.id, role_id=super_user_role.id)
|
|
db.session.add(user_role)
|
|
db.session.commit()
|
|
print("SuperUser role assigned to the new user.")
|
|
else:
|
|
print("Failed to find initial tenant for user creation.")
|
|
else:
|
|
print("Users already exist. Skipping user creation.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
initialize_data()
|
|
except IntegrityError:
|
|
print("Error: Integrity constraint violation. Initial data already exists.")
|
|
except Exception as e:
|
|
print(f"An error occurred during initialization: {e}")
|