Docker deployment Realised

scripts/initialize_data.py

@@ -0,0 +1,87 @@
+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}")