passkey/__init__.py

@@ -1,3 +1,4 @@
+from .bootstrap import bootstrap_if_needed, force_bootstrap
 from .sansio import Passkey
 
-__all__ = ["Passkey"]
+__all__ = ["Passkey", "bootstrap_if_needed", "force_bootstrap"]

passkey/bootstrap.py

@@ -6,7 +6,6 @@ including creating default admin user, organization, permissions, and
 generating a reset link for initial admin setup.
 """
 
-import asyncio
 import logging
 from datetime import datetime
 
@@ -40,39 +39,34 @@ async def _create_and_log_admin_reset_link(user_uuid, message, session_type) ->
     return reset_link
 
 
-async def bootstrap_system(
+async def bootstrap_system() -> dict:
-    user_name: str | None = None, org_name: str | None = None
-) -> dict:
     """
     Bootstrap the entire system with default data.
 
-    Args:
-        user_name: Display name for the admin user (default: "Admin")
-        org_name: Display name for the organization (default: "Organization")
-
     Returns:
         dict: Contains information about created entities and reset link
     """
-    # Create permission first - will fail if already exists
+    admin_uuid = uuid7.create()
-    permission = Permission(id="auth/admin", display_name="Admin")
+    org_uuid = uuid7.create()
-    await globals.db.instance.create_permission(permission)
 
     # Create organization
-    org_uuid = uuid7.create()
     org = Org(
         id=str(org_uuid),
         options={
-            "display_name": org_name or "Organization",
+            "display_name": "Organization",
             "created_at": datetime.now().isoformat(),
         },
     )
     await globals.db.instance.create_organization(org)
 
+    # Create permission
+    permission = Permission(id="auth/admin", display_name="Admin")
+    await globals.db.instance.create_permission(permission)
+
     # Create admin user
-    admin_uuid = uuid7.create()
     user = User(
         uuid=admin_uuid,
-        display_name=user_name or "Admin",
+        display_name="Admin",
         org_uuid=org_uuid,
         role="Admin",
         created_at=datetime.now(),
@@ -119,19 +113,8 @@ async def check_admin_credentials() -> bool:
         bool: True if a reset link was created, False if admin already has credentials
     """
     try:
-        # Get permission organizations to find admin users
+        # Find admin users
-        permission_orgs = await globals.db.instance.get_permission_organizations(
+        admin_users = await globals.db.instance.find_users_by_role("Admin")
-            "auth/admin"
-        )
-
-        if not permission_orgs:
-            return False
-
-        # Get users from the first organization with admin permission
-        org_users = await globals.db.instance.get_organization_users(
-            permission_orgs[0].id
-        )
-        admin_users = [user for user, role in org_users if role == "Admin"]
 
         if not admin_users:
             return False
@@ -157,69 +140,64 @@ async def check_admin_credentials() -> bool:
         return False
 
 
-async def bootstrap_if_needed(
+async def bootstrap_if_needed() -> bool:
-    default_admin: str | None = None, default_org: str | None = None
-) -> bool:
     """
     Check if system needs bootstrapping and perform it if necessary.
 
-    Args:
-        default_admin: Display name for the admin user
-        default_org: Display name for the organization
-
     Returns:
         bool: True if bootstrapping was performed, False if system was already set up
     """
     try:
-        # Check if the admin permission exists - if it does, system is already bootstrapped
+        # Check if any users exist
-        await globals.db.instance.get_permission("auth/admin")
+        if await globals.db.instance.has_any_users():
-        # Permission exists, system is already bootstrapped
+            await check_admin_credentials()
-        # Check if admin needs credentials (only for already-bootstrapped systems)
-        admin_needs_reset = await check_admin_credentials()
-        if not admin_needs_reset:
-            # Use the same format as the reset link messages
-            logger.info(
-                ADMIN_RESET_MESSAGE,
-                "ℹ️  System already bootstrapped - no action needed",
-                "Admin user already has credentials",
-            )
             return False
     except Exception:
-        # Permission doesn't exist, need to bootstrap
         pass
 
-    # No admin permission found, need to bootstrap
+    # No users found, need to bootstrap
-    # Bootstrap creates the admin user AND the reset link, so no need to check credentials after
+    await bootstrap_system()
-    await bootstrap_system(default_admin, default_org)
     return True
 
 
+async def force_bootstrap() -> dict:
+    """
+    Force bootstrap the system (useful for testing or resetting).
+
+    Returns:
+        dict: Bootstrap result information
+    """
+    return await bootstrap_system()
+
+
 # CLI interface
 async def main():
     """Main CLI entry point for bootstrapping."""
     import argparse
 
-    # Configure logging for CLI usage
+    from .db.sql import init
-    logging.basicConfig(level=logging.INFO, format="%(message)s", force=True)
 
     parser = argparse.ArgumentParser(
         description="Bootstrap passkey authentication system"
     )
     parser.add_argument(
-        "--user-name",
+        "--force",
-        default=None,
+        action="store_true",
-        help="Name for the admin user (default: Admin)",
+        help="Force bootstrap even if system is already set up",
-    )
-    parser.add_argument(
-        "--org-name",
-        default=None,
-        help="Name for the organization (default: Organization)",
     )
 
     args = parser.parse_args()
 
-    await globals.init(default_admin=args.user_name, default_org=args.org_name)
+    # Initialize database
+    await init()
+
+    if args.force:
+        await force_bootstrap()
+    else:
+        await bootstrap_if_needed()
 
 
 if __name__ == "__main__":
+    import asyncio
+
     asyncio.run(main())

passkey/db/__init__.py

@@ -231,6 +231,15 @@ class DatabaseInterface(ABC):
     ) -> None:
         """Create a new user and their first credential in a transaction."""
 
+    # Bootstrap helpers
+    @abstractmethod
+    async def has_any_users(self) -> bool:
+        """Check if any users exist in the system."""
+
+    @abstractmethod
+    async def find_users_by_role(self, role: str) -> list[User]:
+        """Find all users with a specific role."""
+
 
 __all__ = [
     "User",

passkey/db/sql.py

@@ -643,3 +643,36 @@ class DB(DatabaseInterface):
             current_time = datetime.now()
             stmt = delete(SessionModel).where(SessionModel.expires < current_time)
             await session.execute(stmt)
+
+    # Bootstrap helpers
+    async def has_any_users(self) -> bool:
+        """Check if any users exist in the system."""
+        async with self.session() as session:
+            stmt = select(UserModel).limit(1)
+            result = await session.execute(stmt)
+            user = result.scalar_one_or_none()
+            return user is not None
+
+    async def find_users_by_role(self, role: str) -> list[User]:
+        """Find all users with a specific role."""
+        async with self.session() as session:
+            stmt = select(UserModel).where(UserModel.role == role)
+            result = await session.execute(stmt)
+            user_models = result.scalars().all()
+
+            users = []
+            for user_model in user_models:
+                user = User(
+                    uuid=UUID(bytes=user_model.uuid),
+                    display_name=user_model.display_name,
+                    org_uuid=UUID(bytes=user_model.org_uuid)
+                    if user_model.org_uuid
+                    else None,
+                    role=user_model.role,
+                    created_at=user_model.created_at,
+                    last_seen=user_model.last_seen,
+                    visits=user_model.visits,
+                )
+                users.append(user)
+
+            return users

passkey/globals.py

@@ -27,11 +27,7 @@ class Manager(Generic[T]):
 
 
 async def init(
-    rp_id: str = "localhost",
+    rp_id: str = "localhost", rp_name: str | None = None, origin: str | None = None
-    rp_name: str | None = None,
-    origin: str | None = None,
-    default_admin: str | None = None,
-    default_org: str | None = None,
 ) -> None:
     """Initialize the global database, passkey instance, and bootstrap the system if needed."""
     # Initialize passkey instance with provided parameters
@@ -52,7 +48,7 @@ async def init(
     # Bootstrap system if needed
     from .bootstrap import bootstrap_if_needed
 
-    await bootstrap_if_needed(default_admin, default_org)
+    await bootstrap_if_needed()
 
 
 # Global instances