Python API Reference¶

def init(
    app_name: Optional[str] = None, template: str = "fastapi_hello", force: bool = False
) -> Dict[str, Any]:
    """
    Initialize a new SFAI application.

    Args:
        app_name: Name of the application (defaults to directory name)
        template: Template to use for scaffolding (eg: fastapi_hello or flask_hello)
        force: Whether to force reinitialization if already initialized

    Returns:
        Dictionary with initialization status and details
    """
    # Use current directory as target path
    target_path = Path.cwd()

    # Determine app name (use directory name if not provided)
    app_name = app_name or target_path.name

    # Check if already initialized
    context_file = target_path / CONTEXT_DIR / CONTEXT_FILE.name
    context_exists = context_file.exists()

    if context_exists and not force:
        return BaseResponse(
            success=False,
            error=(
                f"App '{app_name}' is already initialized. "
                f"Use force=True to reinitialize."
            ),
            app_name=app_name,
        )

    try:
        # Initialize the app with the specified template
        scaffold_hello_app(target_path, force, template)

        # Update environment and register app
        ctx_mgr = ContextManager()
        ctx_mgr.update_platform(
            platform="local",
            environment="default",
            app_name=app_name,
            values={"app_name": app_name},
        )
        ctx_mgr.register_app(app_name, str(target_path))

        return BaseResponse(
            success=True,
            app_name=app_name,
            message=(
                f"App '{app_name}' initialized successfully with template '{template}'"
            ),
        )
    except (ValueError, Exception) as e:
        return BaseResponse(success=False, error=f"Failed to initialize app: {e!s}")

def deploy(
    path: str = ".",
    platform: Optional[str] = None,
    environment: Optional[str] = None,
    **kwargs: Any,
) -> BaseResponse:
    """
    Deploy an application to the configured environment.

    Args:
        path: Path to the application directory
        platform: Platform to deploy to
        environment: Environment to deploy to (uses active environment if None)
        kwargs: Additional keyword arguments

    Returns:
        Dictionary with deployment status and details
    """
    try:
        ctx_mgr = ContextManager()

        # determine the platform and environment to use
        response = determine_platform_and_environment(platform, environment)
        if not response.success:
            return response

        active_platform = response.platform
        active_environment = response.environment

        # Read the updated context for the determined platform and environment
        context = ctx_mgr.read_context(active_platform, active_environment)
        if not context:
            return BaseResponse(
                success=False,
                error=(
                    f"Failed to read context for platform '{active_platform}' "
                    f"and environment '{active_environment}'"
                ),
            )

        provider = PLATFORM_REGISTRY.get(active_platform)
        if not provider:
            return BaseResponse(
                success=False, error=f"Unsupported provider: {active_platform}"
            )

        result = provider.deploy(context=context, path=path, **kwargs)
        return result.with_update(
            app_name=context.get("app_name"),
            platform=active_platform,
            environment=active_environment,
        )
    except ValueError as e:
        return BaseResponse(success=False, error=f"Context validation error: {e!s}")

def open(
    path: str = "/docs",
    port: int = 8080,
    tunnel: bool = False,
    url: str | None = None,
    platform: Optional[str] = None,
    environment: Optional[str] = None,
) -> BaseResponse:
    """
    Open the current app in the browser.

    Args:
        path: The path to open
        port: The port to open
        tunnel: Whether to use a tunnel
        url: The URL to open
        platform: Platform to open from (optional)
        environment: Environment to open from (defaults to "default")
    Returns:
        A dictionary containing the result of the open operation
    """
    try:
        # Load current context
        ctx_mgr = ContextManager()

        # Determine the platform and environment to use
        response = determine_platform_and_environment(platform, environment)
        if not response.success:
            return response

        # Validate platform provider
        active_platform = response.platform
        active_environment = response.environment

        # Read the updated context for the determined platform and environment
        context = ctx_mgr.read_context(active_platform, active_environment)
        if not context:
            return BaseResponse(
                success=False,
                error=(
                    f"Failed to read context for platform '{active_platform}' "
                    f"and environment '{active_environment}'"
                ),
            )

        provider = PLATFORM_REGISTRY.get(active_platform)
        if not provider:
            return BaseResponse(
                success=False, error=f"Unsupported provider: {active_platform}"
            )

        # Validate tunnel requirements
        if tunnel and active_platform != "local":
            return BaseResponse(
                success=False,
                error="Tunneling is only supported in local environment.",
            )

        # Execute the open operation
        result = provider.open(context=context, path=path, url=url)

        # Return result with context metadata
        return result.with_update(
            app_name=context.get("app_name"),
            platform=active_platform,
            environment=active_environment,
        )

    except ValueError as e:
        return BaseResponse(success=False, error=f"Context validation error: {e!s}")

def status(
    platform: Optional[str] = None, environment: Optional[str] = None
) -> BaseResponse:
    """
    Show the status of the current app from context.

    Args:
        platform: Platform to get status from (optional)
        environment: Environment to get status from (uses active environment if None)
    """
    try:
        ctx_mgr = ContextManager()

        # determine the platform and environment to use
        response = determine_platform_and_environment(platform, environment)
        if not response.success:
            return response

        active_platform = response.platform
        active_environment = response.environment

        # Read the updated context for the determined platform and environment
        context = ctx_mgr.read_context(active_platform, active_environment)
        if not context:
            return BaseResponse(
                success=False,
                error=(
                    f"Failed to read context for platform '{active_platform}' "
                    f"and environment '{active_environment}'"
                ),
            )

        provider = PLATFORM_REGISTRY.get(active_platform)
        if not provider:
            return BaseResponse(
                success=False, error=f"Unsupported provider: {active_platform}"
            )
        status_response = provider.status(context=context)
        return status_response.with_update(
            app_name=context.get("app_name"),
            platform=active_platform,
            environment=active_environment,
        )
    except ValueError as e:
        return BaseResponse(success=False, error=f"Context validation error: {e!s}")

def logs(
    platform: Optional[str] = None, environment: Optional[str] = None
) -> BaseResponse:
    """
    Show logs for the current app.

    Args:
        platform: Platform to show logs for
        environment: Environment to show logs for (uses active environment if None)

    Returns:
        BaseResponse: Response object containing logs
    """
    try:
        ctx_mgr = ContextManager()

        # Determine which platform and environment to use
        response = determine_platform_and_environment(platform, environment)
        if not response.success:
            return response

        active_platform = response.platform
        active_environment = response.environment

        # Read the updated context for the determined platform and environment
        context = ctx_mgr.read_context(active_platform, active_environment)
        if not context:
            return BaseResponse(
                success=False,
                error=(
                    f"Failed to read context for platform '{active_platform}' "
                    f"and environment '{active_environment}'"
                ),
            )

        provider = PLATFORM_REGISTRY.get(active_platform)

        if not provider:
            return BaseResponse(
                success=False, error=f"Unsupported provider: {active_platform}"
            )

        logs_response = provider.logs(context=context)

        return logs_response.with_update(
            app_name=context.get("app_name"),
            platform=active_platform,
            environment=active_environment,
        )
    except ValueError as e:
        return BaseResponse(success=False, error=f"Context validation error: {e!s}")

def delete(
    platform: Optional[str] = None, environment: Optional[str] = None
) -> BaseResponse:
    """
    Delete the current app from the context.

    Args:
        platform: Platform to delete from (optional)
        environment: Environment to delete from (uses active environment if None)
    """
    try:
        ctx_mgr = ContextManager()

        # determine the platform and environment to use
        response = determine_platform_and_environment(platform, environment)
        if not response.success:
            return response

        active_platform = response.platform
        active_environment = response.environment

        # Read the updated context for the determined platform and environment
        context = ctx_mgr.read_context(active_platform, active_environment)
        if not context:
            return BaseResponse(
                success=False,
                error=(
                    f"Failed to read context for platform '{active_platform}' "
                    f"and environment '{active_environment}'"
                ),
            )
        provider = PLATFORM_REGISTRY.get(active_platform)

        if not provider:
            return BaseResponse(
                success=False, error=f"Unsupported provider: {active_platform}"
            )

        delete_response = provider.delete(context=context)
        return delete_response.with_update(
            app_name=context.get("app_name"),
            platform=active_platform,
            environment=active_environment,
        )
    except ValueError as e:
        return BaseResponse(success=False, error=f"Context validation error: {e!s}")

def init(
    service: str,
    config: Optional[Dict[str, Any]] = None,
    profile_name: str = "default",
    env: Optional[str] = None,
    **kwargs,
) -> BaseResponse:
    """
    Initialize configuration for a service.

    Args:
        service: Service name (e.g., "mulesoft")
        credentials: Dictionary of credentials for the service
        profile_name: Profile name to store credentials under
        env: Environment to set as active (uses current if None)
        **kwargs: pass values directly to the service provider

    Returns:
        BaseResponse
    """
    ctx_mgr = ContextManager()

    # Read current context to get app info if needed
    context = ctx_mgr.read_context()
    if not context:
        return BaseResponse(
            success=False, error="No app context found. Initialize an app first."
        )

    app_name = context.get("app_name")
    current_platform = env or context.get("active_platform")

    if config is None and not kwargs:
        return BaseResponse(success=False, error="No configuration provided.")
    config = config or kwargs
    if not app_name or not current_platform:
        return BaseResponse(
            success=False, error="App name or platform missing in context."
        )

    # Validate credentials
    result = validate(service, config)
    if not result.success:
        return result.with_update(
            service=service,
            profile_name=profile_name,
        )

    try:
        # Save the service profile
        ctx_mgr.add_service_profile(service, profile_name, config)

        # Update the environment to use this profile
        ctx_mgr.update_platform(current_platform, {service: {"profile": profile_name}})

        return BaseResponse(
            success=True,
            service=service,
            profile_name=profile_name,
        )
    except ValueError as e:
        return BaseResponse(success=False, error=f"Failed to save credentials: {e!s}")

def update(
    service: str, updates: Dict[str, Any], profile_name: str = "default"
) -> BaseResponse:
    """
    Update configuration for a service.

    Args:
        service: Service name
        updates: New values to update
        profile_name: Profile name to update

    Returns:
        BaseResponse
    """
    ctx_mgr = ContextManager()

    # Check if service and profile are valid
    result = profile_exists(service, profile_name)
    if not result.success:
        return result.with_update(
            service=service,
            profile_name=profile_name,
        )

    try:
        ctx_mgr.update_service_profile(service, profile_name, updates)
        return BaseResponse(
            success=True,
            service=service,
            profile_name=profile_name,
            updated_fields=list(updates.keys()),
        )
    except ValueError as e:
        return BaseResponse(
            success=False, error=f"Failed to update configuration: {e!s}"
        )

def list(service: Optional[str] = None) -> BaseResponse:
    """
    List all profiles for one or all services.

    Args:
        service: Service name, or None to list all services

    Returns:
        BaseResponse with operation result and profiles list
    """
    ctx_mgr = ContextManager()

    try:
        if service:
            # Use the ContextManager method for single service
            profiles = ctx_mgr.list_service_profiles(service)
            if not profiles:
                return BaseResponse(
                    success=False, error=f"No profiles found for service: {service}"
                )
            return BaseResponse(success=True, service=service, profiles=profiles)
        else:
            # Get all services and their profiles
            context = ctx_mgr._load_json(ctx_mgr.global_context_file)
            all_profiles = context.get("service_profiles", {})

            if not all_profiles:
                return BaseResponse(success=True, profiles={})

            result = {}
            for svc in all_profiles.keys():
                result[svc] = ctx_mgr.list_service_profiles(svc)

            return BaseResponse(success=True, profiles=result)
    except Exception as e:
        return BaseResponse(success=False, error=f"Failed to list profiles: {e!s}")

def view(service: str, profile_name: str = "default") -> BaseResponse:
    """
    View details of a service profile.

    Args:
        service: Service name
        profile_name: Profile name to view

    Returns:
        BaseResponse
    """
    ctx_mgr = ContextManager()

    # Check if profile exists
    exists, error = profile_exists(service, profile_name)
    if not exists:
        return BaseResponse(
            success=False,
            error=error,
            service=service,
            profile_name=profile_name,
        )

    # Get the profile
    profile = ctx_mgr.get_service_profile(service, profile_name)

    # Create a safe version with masked secrets
    safe_profile = {}
    for key, value in profile.items():
        if any(s in key.lower() for s in ["secret", "password", "key"]):
            # Mask the secret value, showing just the first and last character
            safe_profile[key] = (
                f"{value[0]}{'*' * (len(str(value)) - 2)}{value[-1]}"
                if len(str(value)) > 4
                else "****"
            )
        else:
            safe_profile[key] = value
    return BaseResponse(
        success=True,
        message=f"Profile '{profile_name}' for service '{service}' is shown below",
        profile_name=profile_name,
        service=service,
        data=safe_profile,
    )

def delete(service: str, profile_name: str = "default") -> BaseResponse:
    """
    Delete a service profile.

    Args:
        service: Service name
        profile_name: Profile name to delete

    Returns:
        BaseResponse
    """
    ctx_mgr = ContextManager()

    # Check if profile exists
    exists, error = profile_exists(service, profile_name)
    if not exists:
        return BaseResponse(success=False, error=error)

    try:
        # Delete the profile
        ctx_mgr.delete_service_profile(service, profile_name)

        # Check if this profile is referenced in any environments
        context = ctx_mgr._load_json(ctx_mgr.context_file)
        environments = context.get("environments", {})

        # For each environment, check if it uses this profile
        for env_name, env_data in environments.items():
            if service in env_data and env_data[service].get("profile") == profile_name:
                # Remove the profile reference from this environment
                ctx_mgr.clear_platform_keys(env_name, [service])

        return BaseResponse(
            success=True,
            service=service,
            profile_name=profile_name,
            message=(
                f"Profile '{profile_name}' for service '{service}' has been deleted."
            ),
        )
    except Exception as e:
        return BaseResponse(success=False, error=f"Failed to delete profile: {e!s}")