acme_portal_sdk 1.2.0

acme-portal-sdk helps to manage deployments for applications that implement "flows" (Jobs/DAGs/Workflows). It's intended to be used with acme-portal VSCode extension.

acme-portal-sdk

Important: This SDK is currently in alpha and primarily for demonstration purposes. APIs may still change frequently.

Overview

acme-portal-sdk is a Python SDK that provides data and actions for the acme-portal VSCode extension. It standardizes the deployment workflow for Python applications that implement "flows" (Jobs/DAGs/Workflows) while allowing full customization of the underlying implementation.

AI wiki

Main Idea

Rather than embedding pre-defined deployment logic in the VSCode extension, the SDK allows you to define custom sources of data and behavior. The extension serves as a UI layer to your SDK implementations, providing a consistent interface for:

• Discovering flows in your codebase
• Managing deployments across environments
• Promoting deployments between environments (dev → staging → prod)

The SDK defines abstract interfaces that you implement according to your project's needs, whether using Prefect, Airflow, GitHub Actions, or custom deployment systems.

Quick Start

To set up your project with acme-portal-sdk, create a .acme_portal_sdk directory in your project root with these files:

1. Install the SDK

pip install acme_portal_sdk

2. Create SDK Configuration Files

mkdir .acme_portal_sdk

.acme_portal_sdk/flow_finder.py

from acme_portal_sdk.flow_finder import FlowFinder, FlowDetails
from pathlib import Path
from typing import List
import ast
import os

class MyCustomFlowFinder(FlowFinder):
    """Custom implementation to find flows in your codebase."""
    
    def __init__(self, root_dir: str):
        self.root_dir = Path(root_dir)
    
    def find_flows(self) -> List[FlowDetails]:
        """Find flows by scanning Python files for flow decorators."""
        flows = []
        
        for py_file in self.root_dir.rglob("*.py"):
            if py_file.is_file():
                try:
                    with open(py_file, 'r') as f:
                        tree = ast.parse(f.read())
                    
                    for node in ast.walk(tree):
                        if isinstance(node, ast.FunctionDef):
                            # Look for @flow decorator or similar patterns
                            for decorator in node.decorator_list:
                                if isinstance(decorator, ast.Name) and decorator.id == 'flow':
                                    flows.append(FlowDetails(
                                        name=node.name,
                                        original_name=node.name,
                                        description=ast.get_docstring(node) or "",
                                        obj_type="function",
                                        obj_name=node.name,
                                        obj_parent_type="module",
                                        obj_parent=py_file.stem,
                                        id=f"{py_file.stem}.{node.name}",
                                        module=py_file.stem,
                                        source_path=str(py_file),
                                        source_relative=str(py_file.relative_to(self.root_dir)),
                                        import_path=f"{py_file.stem}",
                                        grouping=[py_file.parent.name]
                                    ))
                except Exception:
                    continue
        
        return flows

# Create an instance to find flows in your project
project_root = Path(__file__).parent.parent
flow_finder = MyCustomFlowFinder(
    root_dir=str(project_root / "src" / "your_project_name")
)

.acme_portal_sdk/deployment_finder.py

from acme_portal_sdk.deployment_finder import DeploymentFinder, DeploymentDetails
from typing import List
import requests

class MyCustomDeploymentFinder(DeploymentFinder):
    """Custom implementation to find existing deployments."""
    
    def __init__(self, api_base_url: str = "https://your-deployment-api.com"):
        self.api_base_url = api_base_url
    
    def get_deployments(self) -> List[DeploymentDetails]:
        """Fetch deployments from your deployment API."""
        try:
            response = requests.get(f"{self.api_base_url}/deployments")
            response.raise_for_status()
            
            deployments = []
            for data in response.json():
                deployments.append(DeploymentDetails(
                    name=data["name"],
                    project_name=data["project"],
                    branch=data["branch"],
                    flow_name=data["flow_name"],
                    env=data["environment"],
                    commit_hash=data["commit_hash"],
                    package_version=data.get("version", "unknown"),
                    tags=data.get("tags", []),
                    id=data["id"],
                    created_at=data["created_at"],
                    updated_at=data["updated_at"],
                    flow_id=data["flow_id"],
                    url=data["url"]
                ))
            
            return deployments
        except Exception:
            return []

# Find existing deployments from your API
deployment_finder = MyCustomDeploymentFinder()

.acme_portal_sdk/flow_deploy.py

from acme_portal_sdk.flow_deploy import DeployWorkflow
from typing import Any, Optional
import subprocess

class MyCustomDeployWorkflow(DeployWorkflow):
    """Custom implementation for deploying flows."""
    
    def __init__(self, deploy_script: str = "scripts/deploy.sh"):
        self.deploy_script = deploy_script
    
    def run(self, *args: Any, **kwargs: Any) -> Optional[str]:
        """Deploy flows using custom deployment script."""
        try:
            flows_to_deploy = kwargs.get("flows_to_deploy", [])
            env = kwargs.get("env", "dev")
            branch_name = kwargs.get("branch_name", "main")
            
            # Build deployment command
            cmd = [
                "bash", self.deploy_script,
                "--flows", ",".join(flows_to_deploy),
                "--env", env,
                "--branch", branch_name
            ]
            
            # Execute deployment
            result = subprocess.run(cmd, capture_output=True, text=True)
            
            if result.returncode == 0:
                # Return deployment URL if available in output
                return result.stdout.strip()
            else:
                print(f"Deployment failed: {result.stderr}")
                return None
                
        except Exception as e:
            print(f"Deployment error: {e}")
            return None

# Deploy flows using custom script
deploy = MyCustomDeployWorkflow(deploy_script="scripts/deploy.sh")

.acme_portal_sdk/deployment_promote.py

from acme_portal_sdk.deployment_promote import PromoteWorkflow
from typing import Any, Optional
import subprocess

class MyCustomPromoteWorkflow(PromoteWorkflow):
    """Custom implementation for promoting deployments between environments."""
    
    def __init__(self, promote_script: str = "scripts/promote.sh"):
        self.promote_script = promote_script
    
    def run(self, *args: Any, **kwargs: Any) -> Optional[str]:
        """Promote deployments using custom promotion script."""
        try:
            flows_to_deploy = kwargs.get("flows_to_deploy", [])
            source_env = kwargs.get("source_env", "dev")
            target_env = kwargs.get("target_env", "prod")
            project_name = kwargs.get("project_name", "")
            
            # Build promotion command
            cmd = [
                "bash", self.promote_script,
                "--flows", ",".join(flows_to_deploy),
                "--from", source_env,
                "--to", target_env,
                "--project", project_name
            ]
            
            # Execute promotion
            result = subprocess.run(cmd, capture_output=True, text=True)
            
            if result.returncode == 0:
                # Return promotion URL if available in output
                return result.stdout.strip()
            else:
                print(f"Promotion failed: {result.stderr}")
                return None
                
        except Exception as e:
            print(f"Promotion error: {e}")
            return None

# Promote deployments using custom script
promote = MyCustomPromoteWorkflow(promote_script="scripts/promote.sh")

3. Install VSCode Extension

Install the acme-portal VSCode extension to get the UI interface for managing your flows and deployments.

Documentation

• Core Concepts - Understanding flows, deployments, and environments
• User Guides - Detailed configuration examples
• Features - Available functionality and platform support
• API Reference - Complete API documentation

Example Projects

• acme-prefect - Complete example using Prefect workflows

Development

For detailed development setup, contribution guidelines, and release notes process, see CONTRIBUTING.md.