API Reference

Base classes

acme_portal_sdk.flow_finder

FlowDetails dataclass

Holds details about a flow.

Flow (often named Workflow/Job/DAG) is a unit of work in a program.

Attributes:

Name	Type	Description
name	str	Display name, may be a normalized version of the original name
original_name	str	Name as defined in the code
description	str	Description of the flow
id	str	Unique identifier for the flow definition in memory
source_path	str	Unambiguous path to the source file from the root of the project
source_relative	str	Relative path to the source file from some known root
line_number	Optional[int]	Line number where the flow is defined in the source file
grouping	List[str]	Desired grouping of the flow in the context of the project (for navigation)
child_attributes	Dict[str, Any]	Additional attributes specific to implementation (e.g., obj_type, obj_name, obj_parent_type, obj_parent, module, import_path for Prefect). Should not be set by subclasses, but may be set by users to add custom information.

from_dict(data: Dict[str, Any]) -> FlowDetails classmethod

Create a FlowDetails instance from a dictionary representation.

to_dict() -> Dict[str, Any]

Convert the FlowDetails to a dictionary suitable for JSON serialization.

FlowFinder

Bases: ABC

Finds flows (units of work/programs) in a given context, with implementations providing specific discovery mechanisms.

find_flows() -> List[FlowDetails] abstractmethod

Method to find flows, to be implemented by subclasses.

acme_portal_sdk.deployment_finder

DeploymentDetails dataclass

Holds details about an existing deployment.

Attributes:

Name	Type	Description
name	str	Original name of the deployment config in the deployment system
project_name	str	Project name to which the deployment belongs to (repo name)
branch	str	Branch of code which is run in the deployment
flow_name	str	Name of the flow run in the deployment
env	str	Environment/Namespace for which the deployment is run (e.g., dev, prod)
commit_hash	str	Commit hash of the code in the deployment
package_version	str	Package version of the code in the deployment
tags	List[str]	Tags associated with the deployment
id	str	Unique identifier for the deployment in the end system
created_at	str	Timestamp of when the deployment was created
updated_at	str	Timestamp of when the deployment was last updated
flow_id	str	Unique identifier for the flow in the deployment system
url	str	URL to the deployment in the deployment system
child_attributes	Dict[str, Any]	Additional attributes that can be set by subclasses

from_dict(data: Dict[str, Any]) -> DeploymentDetails classmethod

Create a DeploymentDetails instance from a dictionary representation.

to_dict() -> Dict[str, Any]

Convert the DeploymentDetails to a dictionary suitable for JSON serialization.

DeploymentFinder

Bases: ABC

Discovers existing deployments in target environments, with implementations providing environment-specific discovery.

get_deployments() -> List[DeploymentDetails] abstractmethod

Method to find deployments, to be implemented by subclasses.

acme_portal_sdk.flow_deploy

DeployInfo dataclass

Holds configuration required to deploy a flow.

Flow deployment contains configuration that defines how to run a unit of work (flow).

Attributes:

Name	Type	Description
name	str	Name of the deployment
flow_name	str	Normalized name of the flow to deploy (from FlowDetails.name)
work_pool_name	Optional[str]	Name of the work pool to use for the deployment, controls which group of resources is used to execute the flow run
work_queue_name	Optional[str]	Name of the work queue to use for the deployment, controls priority of using resources in the pool
parameters	Optional[Dict[str, Any]]	Parameters to pass to the flow when it runs
job_variables	Optional[Dict]	Variables to pass to the job when it runs
cron	Optional[str]	Cron schedule for the deployment
paused	Optional[bool]	Whether the deployment is in an inactive state
concurrency_limit	Optional[int]	Controls possible number of concurrent flow runs
description	Optional[str]	Description of the deployment, overriding the flow description
tags	Optional[List[str]]	Tags to associate with the deployment, for categorization and filtering

DeployInfoPrep

Bases: ABC

Responsible for preparing and validating deployment information before a flow is deployed to a target environment.

prep_deploy_info(*args: List[Any], **kwargs: Dict[Any, Any]) -> List[DeployInfo] abstractmethod

Prepare all deployment information needed to deploy.

Parameters:

Name	Type	Description	Default
*args	List[Any]	Positional arguments	()
**kwargs	Dict[Any, Any]	Keyword arguments	{}

Returns:

Type	Description
List[DeployInfo]	List[DeployInfo]: A list of DeployInfo objects with the prepared deployment information

DeployWorkflow

Bases: ABC

Encapsulates the deployment workflow for a flow.

__call__(*args: List[Any], **kwargs: Dict[Any, Any]) -> Optional[str]

Call the run method with the provided arguments.

Parameters:

Name	Type	Description	Default
**args	List[Any]	Positional arguments	()
**kwargs	Dict[Any, Any]	Keyword arguments	{}

Returns:

Type	Description
Optional[str]	Optional[str]: URL of the deployment if successful, None otherwise

run(*args: Any, **kwargs: Any) -> Optional[str] abstractmethod

Run the deployment workflow for the specified flows.

This method accepts flexible arguments to allow different implementations to require different parameters based on their specific needs.

Common parameters that implementations may expect

flows_to_deploy: List of flow names to deploy ref: The git ref (branch/tag) for the workflow project_name: Name of the project branch_name: Name of the branch commit_hash: Git commit hash image_uri: Docker image URI package_version: Package version env: Target environment

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments specific to the implementation	()
**kwargs	Any	Keyword arguments specific to the implementation	{}

Returns:

Type	Description
Optional[str]	Optional[str]: URL of the deployment if successful, None otherwise

FlowDeployer

Bases: ABC

Deploys flows, with implementations handling the deployment to specific execution environment.

deploy(flow_deploy_info: DeployInfo) -> None abstractmethod

Deploy a flow.

Parameters:

Name	Type	Description	Default
flow_deploy_info	DeployInfo	Configuration for the deployment	required

acme_portal_sdk.deployment_promote

DeploymentPromote

Bases: ABC

Responsible for promoting flows between different environments (e.g., dev to prod), managing the transition of deployments

promote(project_name: str, branch_name: str, source_env: str, target_env: str, flows_to_deploy: List[str]) abstractmethod

Promote flows from one environment to another.

Parameters:

Name	Type	Description	Default
project_name	str	Name of the project	required
branch_name	str	Name of the branch	required
source_env	str	Source environment	required
target_env	str	Target environment	required
flows_to_deploy	List[str]	List of flow names to promote	required

PromoteWorkflow

Bases: ABC

Encapsulates the workflow for promoting flows between environments.

__call__(*args: List[Any], **kwargs: Dict[Any, Any]) -> Optional[str]

Call the run method with the provided arguments.

Parameters:

Name	Type	Description	Default
**args	List[Any]	Positional arguments	()
**kwargs	Dict[Any, Any]	Keyword arguments	{}

Returns: Optional[str]: URL of the promotion if successful, None otherwise

run(*args: Any, **kwargs: Any) -> Optional[str] abstractmethod

Run the promotion workflow for the specified flows.

This method accepts flexible arguments to allow different implementations to require different parameters based on their specific needs.

Common parameters that implementations may expect

flows_to_deploy: List of flow names to promote source_env: Source environment target_env: Target environment ref: The git ref (branch/tag) for the workflow project_name: Name of the project branch_name: Name of the branch

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments specific to the implementation	()
**kwargs	Any	Keyword arguments specific to the implementation	{}

Returns:

Type	Description
Optional[str]	Optional[str]: URL of the promotion if successful, None otherwise

Prefect

acme_portal_sdk.prefect.flow_finder

PrefectFlowAttributes dataclass

Dataclass to capture Prefect-specific attributes stored in child_attributes.

This dataclass represents the implementation-specific metadata that Prefect flow finder collects about discovered flows. These attributes provide detailed information about how the flow is implemented in Python code.

Attributes:

Name	Type	Description
obj_type	str	Type of object defining the flow ("function" or "method")
obj_name	str	Name of the function or method that defines the flow
obj_parent_type	str	Type of container for the flow object ("module" or "class")
obj_parent	str	Name of the module or class containing the flow
module	str	Python module name where the flow is defined
import_path	str	Full Python import path to the source file

to_dict() -> Dict[str, str]

Convert the dataclass to a dictionary for use in child_attributes.

PrefectFlowFinder

Bases: FlowFinder

Scans Python code directories to identify Prefect flows by analyzing decorators, extracting metadata and organizing found flows into flat list.

acme_portal_sdk.prefect.deployment_finder

PrefectDeploymentFinder

Bases: DeploymentFinder

Finds Prefect deployments in a given context.

Connects to Prefect's API to discover and retrieve information about existing deployments in the Prefect backend.

__init__()

Initialize the PrefectDeploymentFinder and verify Prefect credentials.

get_deployments() -> List[DeploymentDetails]

Connect to Prefect and get deployment information.

acme_portal_sdk.prefect.flow_deploy

PrefectDeployInfo dataclass

Bases: DeployInfo

Dataclass that extends DeployInfo with Prefect-specific deployment configuration

Attributes:

Name	Type	Description
triggers	Optional[Dict[str, Dict[Any, Any]]]	Each key in the dictionary identifies the trigger class and value provides the parameters for the trigger type
image_uri	Optional[str]	Docker image URI for the flow deployment
version	Optional[str]	Version identifier for the deployment
flow_function	Optional[Any]	The actual flow function object

get_trigger_instances() -> List[Any]

Convert the triggers dictionary into a list of trigger class instances.

Returns:

Type	Description
List[Any]	List[Any]: A list of instantiated trigger objects

PrefectDeployInfoPrep

Bases: DeployInfoPrep

Prepares deployment information from a configuration file and runtime context and standardizes deployment parameters.

prep_deploy_info(project_name: str, branch_name: str, commit_hash: str, image_uri: str, package_version: str, env: str, flows_to_deploy: List[str], env_vars: Optional[Dict[str, str]] = None) -> List[PrefectDeployInfo]

Prepare deployment information from provided context.

Parameters:

Name	Type	Description	Default
project_name	str	Name of the project	required
branch_name	str	Name of the branch	required
commit_hash	str	Git commit hash	required
image_uri	str	Docker image URI	required
package_version	str	Package version	required
env	str	Environment for deployment	required
flows_to_deploy	List[str]	List of flow names to deploy	required
env_vars	Optional[Dict[str, str]]	Additional environment variables for the deployment	None

Returns:

Type	Description
List[PrefectDeployInfo]	List[PrefectDeployInfo]: List of prepared deployment information objects

PrefectFlowDeployer

Bases: FlowDeployer

Uses the deployment info for desired flows to invoke Prefect flow deployment.

deploy(flow_deploy_info: PrefectDeployInfo) -> None

Deploy a Prefect flow using the provided configuration.

Parameters:

Name	Type	Description	Default
flow_deploy_info	PrefectDeployInfo	Configuration for the deployment including the flow function	required

acme_portal_sdk.prefect.deployment_promote

PrefectDeploymentPromote

Bases: DeploymentPromote

Promotes Prefect deployments from one Environment to another.

__init__(deployer: PrefectFlowDeployer, flow_deploy_info_prep: PrefectDeployInfoPrep)

Initialize the DeploymentPromote with a flow deployer.

Parameters:

Name	Type	Description	Default
deployer	PrefectFlowDeployer	A flow deployer instance to handle the actual deployment	required
flow_deploy_info_prep	PrefectDeployInfoPrep	A helper to prepare deployment info	required

promote(project_name: str, branch_name: str, source_env: str, target_env: str, flows_to_deploy: List[str], target_env_vars: Optional[Dict[str, str]] = None) -> None

Promote flows from one environment to another.

Parameters:

Name	Type	Description	Default
project_name	str	Name of the project	required
branch_name	str	Name of the branch	required
source_env	str	Source environment	required
target_env	str	Target environment	required
flows_to_deploy	List[str]	List of flow names to promote	required
target_env_vars	Optional[Dict[str, str]]	Optional environment variables to set for the target environment	None

GitHub

acme_portal_sdk.github.github_workflow

CommandExecutor

Executes shell commands and returns their output.

execute(command: str, working_dir: Optional[str] = None) -> Tuple[str, str]

Execute a shell command and return its output.

Parameters:

Name	Type	Description	Default
command	str	The command to execute	required
working_dir	Optional[str]	The directory to run the command in	None

Returns:

Type	Description
Tuple[str, str]	Tuple of (stdout, stderr) from the command

GitHubWorkflowService

Service for GitHub workflow operations.

parse_workflow_run_response(run_list_output: str) -> Optional[str]

Parse the GitHub CLI response for workflow run information.

Parameters:

Name	Type	Description	Default
run_list_output	str	The output from the GitHub CLI run list command	required

Returns:

Type	Description
Optional[str]	URL to the workflow run or None

trigger_workflow(workflow_file: str, workflow_inputs: Dict[str, str], workflow_ref: str = 'main') -> Optional[str]

Trigger a GitHub workflow.

Parameters:

Name	Type	Description	Default
workflow_file	str	The workflow file name	required
workflow_inputs	Dict[str, str]	The inputs for the workflow	required
workflow_ref	str	The git ref (branch/tag) for the workflow	'main'

Returns:

Type	Description
Optional[str]	URL to the workflow run or None on error

trigger_workflow_via_github_cli(workflow_file: str, workflow_inputs: Dict[str, str], workflow_ref: str) -> Optional[str]

Trigger a GitHub workflow using the GitHub CLI.

Parameters:

Name	Type	Description	Default
workflow_file	str	The workflow file name	required
workflow_inputs	Dict[str, str]	The inputs for the workflow	required
workflow_ref	str	The git ref (branch/tag) for the workflow	required

Returns:

Type	Description
Optional[str]	URL to the workflow run or None on error

GitService

Provides Git-related functionality.

get_repository_url(working_dir: Optional[str] = None) -> Optional[str]

Get the URL of the Git repository.

Parameters:

Name	Type	Description	Default
working_dir	Optional[str]	The directory to run the Git command in	None

Returns:

Type	Description
Optional[str]	The URL of the repository or None if not found

GithubActionsDeployWorkflow

Bases: DeployWorkflow

Implements the DeployWorkflow interface using GitHub Actions

__init__(workflow_file: str = 'deploy.yml', default_ref: str = 'main')

Initialize the GitHub Actions deploy workflow.

Parameters:

Name	Type	Description	Default
workflow_file	str	The workflow file name (default: deploy.yml)	'deploy.yml'
default_ref	str	Default git ref to use if none provided in run() (default: main)	'main'

run(flows_to_deploy: List[str], ref: Optional[str] = None, **kwargs) -> Optional[str]

Run the deployment workflow for the specified flows using GitHub Actions.

Parameters:

Name	Type	Description	Default
flows_to_deploy	List[str]	List of flow names to deploy	required
ref	Optional[str]	The git ref (branch/tag) for the workflow (optional, uses default_ref if not provided)	None
**kwargs		Additional workflow parameters (for future extensibility)	{}

Returns:

Type	Description
Optional[str]	Optional[str]: URL of the workflow run if successful, None otherwise

GithubActionsPromoteWorkflow

Bases: PromoteWorkflow

Implements the PromoteWorkflow interface using GitHub Actions.

__init__(workflow_file: str = 'promote.yml')

Initialize the GitHub Actions promote workflow.

Parameters:

Name	Type	Description	Default
workflow_file	str	The workflow file name (default: promote.yml)	'promote.yml'

run(flows_to_deploy: List[str], source_env: Optional[str] = None, target_env: Optional[str] = None, ref: Optional[str] = None, **kwargs) -> Optional[str]

Run the promotion workflow for the specified flows using GitHub Actions.

Parameters:

Name	Type	Description	Default
flows_to_deploy	List[str]	List of flow names to promote	required
source_env	Optional[str]	Source environment (required)	None
target_env	Optional[str]	Target environment (required)	None
ref	Optional[str]	The git ref (branch/tag) for the workflow (required)	None
**kwargs		Additional workflow parameters (for future extensibility)	{}

Returns:

Type	Description
Optional[str]	Optional[str]: URL of the workflow run if successful, None otherwise