357629b1-60cb-43d6-9ac2-8b5249dd574b

Committed 26 Mar 2026 02:28PM UTC coverage: 36.086% (-0.3%) from 36.378%

Build # 357629b1-60cb-43d6-9ac2-8b5249dd574b

Build Type

Pull #2060

circleci

Committed by

jlaneve

Commit Message

fix: add include/ to PYTHONPATH in standalone mode

Standalone mode's buildEnv() did not set PYTHONPATH to include the
project's include/ directory. In Docker mode the Astronomer runtime
image bakes /usr/local/airflow/include into PYTHONPATH, but standalone
had no equivalent — causing import failures for any user modules in
include/ during pytest, parse, and start.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Pull Request Pull Request #2060: fix: add include/ to PYTHONPATH in standalone mode

Coverage Stats

7 of 8 new or added lines in 1 file covered. (87.5%)

403 existing lines in 8 files now uncovered.

24494 of 67876 relevant lines covered (36.09%)

8.63 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.19

/cmd/cloud/deployment_objects.go

package cloud

import (
        "fmt"
        "io"
        "strings"

        airflowversions "github.com/astronomer/astro-cli/airflow_versions"
        astroplatformcore "github.com/astronomer/astro-cli/astro-client-platform-core"
        "github.com/astronomer/astro-cli/cloud/deployment"
        "github.com/astronomer/astro-cli/cloud/deployment/inspect"
        "github.com/pkg/errors"
        "github.com/spf13/cobra"
)

var (
        connID             string
        connType           string
        host               string
        login              string
        password           string
        schema             string
        extra              string
        fromDeploymentID   string
        fromDeploymentName string
        toDeploymentID     string
        toDeploymentName   string
        port               int
        varValue           string
        key                string
        slots              int
        includeDeferred    string
)

const (
        webserverURLField        = "metadata.airflow_api_url"
        warningConnectionCopyCMD = "WARNING! The password and extra field are not copied over. You will need to manually add these values"
        warningVariableCopyCMD   = "WARNING! Secret values are not copied over. You will need to manually add these values"
)

func newDeploymentConnectionRootCmd(out io.Writer) *cobra.Command {
        cmd := &cobra.Command{
                Use:     "connection",
                Aliases: []string{"con", "connections"},
                Short:   "Manage deployment connections",
                Long:    "Manage connections for an Astro Deployment.",
        }
        cmd.AddCommand(
                newDeploymentConnectionListCmd(out),
                newDeploymentConnectionCreateCmd(out),
                newDeploymentConnectionUpdateCmd(out),
                newDeploymentConnectionCopyCmd(out),
        )
        return cmd
}

func newDeploymentConnectionListCmd(out io.Writer) *cobra.Command {
        cmd := &cobra.Command{
                Use:     "list",
                Aliases: []string{"li"},
                Short:   "list a Deployment's connections",
                Long:    "list connections for an Astro Deployment",
                RunE: func(cmd *cobra.Command, args []string) error {
                        return deploymentConnectionList(cmd, out)
                },
        }
        cmd.Flags().StringVarP(&deploymentID, "deployment-id", "d", "", "The ID of the Deployment.")
        cmd.Flags().StringVarP(&deploymentName, "deployment-name", "n", "", "The name of the Deployment.")

        return cmd
}

//nolint:dupl
func newDeploymentConnectionCreateCmd(out io.Writer) *cobra.Command {
        cmd := &cobra.Command{
                Use:     "create",
                Aliases: []string{"cr"},
                Short:   "Create connections for a Deployment",
                Long:    "Create Airflow connections for an Astro Deployment",
                RunE: func(cmd *cobra.Command, args []string) error {
                        return deploymentConnectionCreate(cmd, out)
                },
        }
        cmd.Flags().StringVarP(&deploymentID, "deployment-id", "d", "", "The ID of the Deployment.")
        cmd.Flags().StringVarP(&deploymentName, "deployment-name", "n", "", "The name of the Deployment.")
        cmd.Flags().StringVarP(&connID, "conn-id", "i", "", "The connection ID. Required.")
        cmd.Flags().StringVarP(&connType, "conn-type", "t", "", "The connection type. Required.")
        cmd.Flags().StringVarP(&description, "description", "", "", "The connection description.")
        cmd.Flags().StringVarP(&host, "host", "", "", "The connection host.")
        cmd.Flags().StringVarP(&login, "login", "l", "", "The connection login or username.")
        cmd.Flags().StringVarP(&password, "password", "p", "", "The connection password.")
        cmd.Flags().StringVarP(&schema, "schema", "s", "", "The connection schema.")
        cmd.Flags().IntVarP(&port, "port", "o", 0, "The connection port.")
        cmd.Flags().StringVarP(&extra, "extra", "e", "", "The extra field configuration, defined as a stringified JSON object.")

        return cmd
}

//nolint:dupl
func newDeploymentConnectionUpdateCmd(out io.Writer) *cobra.Command {
        cmd := &cobra.Command{
                Use:     "update",
                Aliases: []string{"up"},
                Short:   "Update connections for a Deployment",
                Long:    "Update existing Airflow connections for an Astro Deployment",
                RunE: func(cmd *cobra.Command, args []string) error {
                        return deploymentConnectionUpdate(cmd, out)
                },
        }
        cmd.Flags().StringVarP(&deploymentID, "deployment-id", "d", "", "The ID of the Deployment.")
        cmd.Flags().StringVarP(&deploymentName, "deployment-name", "n", "", "The name of the Deployment.")
        cmd.Flags().StringVarP(&connID, "conn-id", "i", "", "The connection ID. Required.")
        cmd.Flags().StringVarP(&connType, "conn-type", "t", "", "The connection type. Required.")
        cmd.Flags().StringVarP(&description, "description", "", "", "The connection description.")
        cmd.Flags().StringVarP(&host, "host", "", "", "The connection host.")
        cmd.Flags().StringVarP(&login, "login", "l", "", "The connection login or username.")
        cmd.Flags().StringVarP(&password, "password", "p", "", "The connection password.")
        cmd.Flags().StringVarP(&schema, "schema", "s", "", "The connection schema.")
        cmd.Flags().IntVarP(&port, "port", "o", 0, "The connection port.")
        cmd.Flags().StringVarP(&extra, "extra", "e", "", "The extra field configuration, defined as a stringified JSON object.")

        return cmd
}

//nolint:dupl
func newDeploymentConnectionCopyCmd(out io.Writer) *cobra.Command {
        cmd := &cobra.Command{
                Use:     "copy",
                Aliases: []string{"cp"},
                Short:   "Copy connections from one Deployment to another",
                Long:    "Copy Airflow connections from one Astro Deployment to another. Passwords and extra configurations will not copy over. If a connection already exits with same connection ID in the target Deployment, that connection will be updated",
                RunE: func(cmd *cobra.Command, args []string) error {
                        return deploymentConnectionCopy(cmd, out)
                },
        }
        cmd.Flags().StringVarP(&fromDeploymentID, "source-id", "s", "", "The ID of the Deployment to copy connections from.")
        cmd.Flags().StringVarP(&fromDeploymentName, "source-name", "n", "", "The name of the Deployment to copy connections from")
        cmd.Flags().StringVarP(&toDeploymentID, "target-id", "t", "", "The ID of the Deployment to receive the copied connections")
        cmd.Flags().StringVarP(&toDeploymentName, "target-name", "", "", "The name of the Deployment to receive the copied connections")

        return cmd
}

func newDeploymentAirflowVariableRootCmd(out io.Writer) *cobra.Command {
        cmd := &cobra.Command{
                Use:     "airflow-variable",
                Aliases: []string{"var"},
                Short:   "Manage Airflow variables in an Astro Deployment",
                Long:    "Manage Airflow variables stored in an Astro Deployment's metadata database.",
        }
        cmd.AddCommand(
                newDeploymentAirflowVariableListCmd(out),
                newDeploymentAirflowVariableCreateCmd(out),
                newDeploymentAirflowVariableUpdateCmd(out),
                newDeploymentAirflowVariableCopyCmd(out),
        )
        return cmd
}

func newDeploymentAirflowVariableListCmd(out io.Writer) *cobra.Command {
        cmd := &cobra.Command{
                Use:     "list",
                Aliases: []string{"li"},
                Short:   "list a Deployment's Airflow variables",
                Long:    "list Airflow variables stored in an Astro Deployment's metadata database",
                RunE: func(cmd *cobra.Command, args []string) error {
                        return deploymentAirflowVariableList(cmd, out)
                },
        }
        cmd.Flags().StringVarP(&deploymentID, "deployment-id", "d", "", "The ID of the Deployment.")
        cmd.Flags().StringVarP(&deploymentName, "deployment-name", "n", "", "The name of the Deployment.")

        return cmd
}

//nolint:dupl
func newDeploymentAirflowVariableCreateCmd(out io.Writer) *cobra.Command {
        cmd := &cobra.Command{
                Use:     "create",
                Aliases: []string{"cr"},
                Short:   "Create Airflow variables for a Deployment",
                Long:    "Create Airflow variables for an Astro Deployment",
                RunE: func(cmd *cobra.Command, args []string) error {
                        return deploymentAirflowVariableCreate(cmd, out)
                },
        }
        cmd.Flags().StringVarP(&deploymentID, "deployment-id", "d", "", "The ID of the Deployment.")
        cmd.Flags().StringVarP(&deploymentName, "deployment-name", "n", "", "The name of the Deployment.")
        cmd.Flags().StringVarP(&varValue, "value", "v", "", "The Airflow variable value. Required.")
        cmd.Flags().StringVarP(&key, "key", "k", "", "The Airflow variable key. Required.")
        cmd.Flags().StringVarP(&description, "description", "", "", "The Airflow variable description.")

        return cmd
}

//nolint:dupl
func newDeploymentAirflowVariableUpdateCmd(out io.Writer) *cobra.Command {
        cmd := &cobra.Command{
                Use:     "update",
                Aliases: []string{"up"},
                Short:   "Update Airflow variables for a Deployment",
                Long:    "Update Airflow variables for an Astro Deployment",
                RunE: func(cmd *cobra.Command, args []string) error {
                        return deploymentAirflowVariableUpdate(cmd, out)
                },
        }
        cmd.Flags().StringVarP(&deploymentID, "deployment-id", "d", "", "The ID of the Deployment.")
        cmd.Flags().StringVarP(&deploymentName, "deployment-name", "n", "", "The name of the Deployment.")
        cmd.Flags().StringVarP(&varValue, "value", "v", "", "The Airflow variable value. Required.")
        cmd.Flags().StringVarP(&key, "key", "k", "", "The Airflow variable key. Required.")
        cmd.Flags().StringVarP(&description, "description", "", "", "The Airflow variable description.")

        return cmd
}

//nolint:dupl
func newDeploymentAirflowVariableCopyCmd(out io.Writer) *cobra.Command {
        cmd := &cobra.Command{
                Use:     "copy",
                Aliases: []string{"cp"},
                Short:   "Copy the Airflow variables from one Deployment to another",
                Long:    "Copy Airflow variables from one Astro Deployment to another Astro Deployment. If a variable already exits with same Key it will be updated",
                RunE: func(cmd *cobra.Command, args []string) error {
                        return deploymentAirflowVariableCopy(cmd, out)
                },
        }
        cmd.Flags().StringVarP(&fromDeploymentID, "source-id", "s", "", "The ID of the Deployment to copy Airflow variables from.")
        cmd.Flags().StringVarP(&fromDeploymentName, "source-name", "n", "", "The name of the Deployment to copy Airflow variables from")
        cmd.Flags().StringVarP(&toDeploymentID, "target-id", "t", "", "The ID of the Deployment to receive the copied Airflow variables")
        cmd.Flags().StringVarP(&toDeploymentName, "target-name", "", "", "The name of the Deployment to receive the copied Airflow variables")

        return cmd
}

func newDeploymentPoolRootCmd(out io.Writer) *cobra.Command {
        cmd := &cobra.Command{
                Use:     "pool",
                Aliases: []string{"pl", "pools"},
                Short:   "Manage Deployment's Airflow pools",
                Long:    "Manage Airflow pools stored for an Astro Deployment",
        }
        cmd.AddCommand(
                newDeploymentPoolListCmd(out),
                newDeploymentPoolCreateCmd(out),
                newDeploymentPoolUpdateCmd(out),
                newDeploymentPoolCopyCmd(out),
        )
        return cmd
}

func newDeploymentPoolListCmd(out io.Writer) *cobra.Command {
        cmd := &cobra.Command{
                Use:     "list",
                Aliases: []string{"li"},
                Short:   "list a Deployment's Airflow pools",
                Long:    "list Airflow pools for an Astro Deployment",
                RunE: func(cmd *cobra.Command, args []string) error {
                        return deploymentPoolList(cmd, out)
                },
        }
        cmd.Flags().StringVarP(&deploymentID, "deployment-id", "d", "", "The ID of the Deployment.")
        cmd.Flags().StringVarP(&deploymentName, "deployment-name", "n", "", "The name of the Deployment.")

        return cmd
}

//nolint:dupl
func newDeploymentPoolCreateCmd(out io.Writer) *cobra.Command {
        cmd := &cobra.Command{
                Use:     "create",
                Aliases: []string{"cr"},
                Short:   "Create Airflow pools for an Astro Deployment",
                Long:    "Create Airflow pools for an Astro Deployment",
                RunE: func(cmd *cobra.Command, args []string) error {
                        return deploymentPoolCreate(cmd, out)
                },
        }
        cmd.Flags().StringVarP(&deploymentID, "deployment-id", "d", "", "The ID of the Deployment.")
        cmd.Flags().StringVarP(&deploymentName, "deployment-name", "n", "", "The name of the Deployment.")
        cmd.Flags().StringVarP(&name, "name", "", "", "The Airflow pool value. Required")
        cmd.Flags().IntVarP(&slots, "slots", "s", 0, "The Airflow pool key. Required")
        cmd.Flags().StringVarP(&description, "description", "", "", "The Airflow pool description")
        cmd.Flags().StringVarP(&includeDeferred, "include-deferred", "", "", "If set to 'enable', deferred tasks are considered when calculating open pool slots. Default is 'disable'. Possible values are disable/enable.")

        return cmd
}

//nolint:dupl
func newDeploymentPoolUpdateCmd(out io.Writer) *cobra.Command {
        cmd := &cobra.Command{
                Use:     "update",
                Aliases: []string{"up"},
                Short:   "Update Airflow pools for an Astro Deployment",
                Long:    "Update Airflow pools for an Astro Deployment",
                RunE: func(cmd *cobra.Command, args []string) error {
                        return deploymentPoolUpdate(cmd, out)
                },
        }
        cmd.Flags().StringVarP(&deploymentID, "deployment-id", "d", "", "The ID of the Deployment.")
        cmd.Flags().StringVarP(&deploymentName, "deployment-name", "n", "", "The name of the Deployment.")
        cmd.Flags().StringVarP(&name, "name", "", "", "The pool value.  Required.")
        cmd.Flags().IntVarP(&slots, "slots", "s", 0, "The pool slots.")
        cmd.Flags().StringVarP(&description, "description", "", "", "The pool description.")
        cmd.Flags().StringVarP(&includeDeferred, "include-deferred", "", "", "If set to 'enable', deferred tasks are considered when calculating open pool slots. Required for Airflow 3+ deployments. Possible values disable/enable.")

        return cmd
}

//nolint:dupl
func newDeploymentPoolCopyCmd(out io.Writer) *cobra.Command {
        cmd := &cobra.Command{
                Use:     "copy",
                Aliases: []string{"cp"},
                Short:   "Copy Airflow pools from one Astro Deployment to another",
                Long:    "Copy Airflow pools from one Astro Deployment to another Astro Deployment. If a pool already exits with same name it will be updated",
                RunE: func(cmd *cobra.Command, args []string) error {
                        return deploymentPoolCopy(cmd, out)
                },
        }
        cmd.Flags().StringVarP(&fromDeploymentID, "source-id", "s", "", "The ID of the Deployment to copy Airflow pools from.")
        cmd.Flags().StringVarP(&fromDeploymentName, "source-name", "n", "", "The name of the Deployment to copy Airflow pools from")
        cmd.Flags().StringVarP(&toDeploymentID, "target-id", "t", "", "The ID of the Deployment to receive the copied Airflow pools")
        cmd.Flags().StringVarP(&toDeploymentName, "target-name", "", "", "The name of the Deployment to receive the copied Airflow pools")

        return cmd
}

//nolint:dupl
func deploymentConnectionList(cmd *cobra.Command, out io.Writer) error {
        ws, err := coalesceWorkspace()
        if err != nil {
                return errors.Wrap(err, "failed to find a valid workspace")
        }

        // Silence Usage as we have now validated command input
        cmd.SilenceUsage = true

        // get or select the deployment
        requestedDeployment, err := deployment.GetDeployment(ws, deploymentID, deploymentName, false, nil, platformCoreClient, astroCoreClient)
        if err != nil {
                return err
        }

        airflowURL, err := getAirflowURL(&requestedDeployment)
        if err != nil {
                return err
        }

        return deployment.ConnectionList(airflowURL, airflowAPIClient, out)
}

func deploymentConnectionCreate(cmd *cobra.Command, out io.Writer) error {
        ws, err := coalesceWorkspace()
        if err != nil {
                return errors.Wrap(err, "failed to find a valid Workspace")
        }

        // Silence Usage as we have now validated command input
        cmd.SilenceUsage = true

        // get or select the deployment
        requestedDeployment, err := deployment.GetDeployment(ws, deploymentID, deploymentName, false, nil, platformCoreClient, astroCoreClient)
        if err != nil {
                return err
        }

        airflowURL, err := getAirflowURL(&requestedDeployment)
        if err != nil {
                return err
        }

        // check connID and connType
        if connID == "" {
                return errors.New("a connection ID is needed to create a connection. Please use the '--conn-id' flag to specify a connection ID")
        }

        if connType == "" {
                return errors.New("a connection type is needed to create a connection. Please use the '--conn-type' flag to specify a connection type")
        }

        return deployment.ConnectionCreate(airflowURL, connID, connType, description, host, login, password, schema, extra, port, airflowAPIClient, out)
}

func deploymentConnectionUpdate(cmd *cobra.Command, out io.Writer) error {
        ws, err := coalesceWorkspace()
        if err != nil {
                return errors.Wrap(err, "failed to find a valid workspace")
        }

        // Silence Usage as we have now validated command input
        cmd.SilenceUsage = true

        // get or select the deployment
        requestedDeployment, err := deployment.GetDeployment(ws, deploymentID, deploymentName, false, nil, platformCoreClient, astroCoreClient)
        if err != nil {
                return err
        }

        airflowURL, err := getAirflowURL(&requestedDeployment)
        if err != nil {
                return err
        }

        // check connID and connType
        if connID == "" {
                return errors.New("a connection ID is needed to create a connection. Please use the '--conn-id' flag to specify a connection ID")
        }

        return deployment.ConnectionUpdate(airflowURL, connID, connType, description, host, login, password, schema, extra, port, airflowAPIClient, out)
}

//nolint:dupl
func deploymentConnectionCopy(cmd *cobra.Command, out io.Writer) error {
        ws, err := coalesceWorkspace()
        if err != nil {
                return errors.Wrap(err, "failed to find a valid Workspace")
        }

        // Silence Usage as we have now validated command input
        cmd.SilenceUsage = true

        // get source deployment
        if fromDeploymentName == "" && fromDeploymentID == "" {
                fmt.Println("Which Deployment should connections be copied from?")
        }
        fromDeployment, err := deployment.GetDeployment(ws, fromDeploymentID, fromDeploymentName, false, nil, platformCoreClient, astroCoreClient)
        if err != nil {
                return errors.Wrap(err, "failed to find the source Deployment")
        }

        fromAirflowURL, err := getAirflowURL(&fromDeployment)
        if err != nil {
                return errors.Wrap(err, "failed to find the source Deployment Airflow webserver URL")
        }

        // get target deployment
        if toDeploymentName == "" && toDeploymentID == "" {
                fmt.Println("Which Deployment should receive the connections?")
        }
        toDeployment, err := deployment.GetDeployment(ws, toDeploymentID, toDeploymentName, false, nil, platformCoreClient, astroCoreClient)
        if err != nil {
                return errors.Wrap(err, "failed to find the target Deployment")
        }

        toAirflowURL, err := getAirflowURL(&toDeployment)
        if err != nil {
                return errors.Wrap(err, "failed to find the target Deployment Airflow webserver URL")
        }

        fmt.Println(warningConnectionCopyCMD)
        return deployment.CopyConnection(fromAirflowURL, toAirflowURL, airflowAPIClient, out)
}

//nolint:dupl
func deploymentAirflowVariableList(cmd *cobra.Command, out io.Writer) error {
        ws, err := coalesceWorkspace()
        if err != nil {
                return errors.Wrap(err, "failed to find a valid Workspace")
        }

        // Silence Usage as we have now validated command input
        cmd.SilenceUsage = true

        // get or select the deployment
        requestedDeployment, err := deployment.GetDeployment(ws, deploymentID, deploymentName, false, nil, platformCoreClient, astroCoreClient)
        if err != nil {
                return err
        }

        airflowURL, err := getAirflowURL(&requestedDeployment)
        if err != nil {
                return err
        }

        return deployment.AirflowVariableList(airflowURL, airflowAPIClient, out)
}

func deploymentAirflowVariableCreate(cmd *cobra.Command, out io.Writer) error {
        ws, err := coalesceWorkspace()
        if err != nil {
                return errors.Wrap(err, "failed to find a valid workspace")
        }

        // Silence Usage as we have now validated command input
        cmd.SilenceUsage = true

        // get or select the deployment
        requestedDeployment, err := deployment.GetDeployment(ws, deploymentID, deploymentName, false, nil, platformCoreClient, astroCoreClient)
        if err != nil {
                return err
        }

        airflowURL, err := getAirflowURL(&requestedDeployment)
        if err != nil {
                return err
        }

        // check key and varValue
        if key == "" {
                return errors.New("a variable key is needed to create an airflow variable. Please use the '--key' flag to specify a key")
        }

        if varValue == "" {
                return errors.New("a variable value is needed to create an airflow variable. Please use the '--value' flag to specify a value")
        }

        return deployment.VariableCreate(airflowURL, varValue, key, description, airflowAPIClient, out)
}

func deploymentAirflowVariableUpdate(cmd *cobra.Command, out io.Writer) error { //nolint
        ws, err := coalesceWorkspace()
        if err != nil {
                return errors.Wrap(err, "failed to find a valid workspace")
        }

        // Silence Usage as we have now validated command input
        cmd.SilenceUsage = true

        // get or select the deployment
        requestedDeployment, err := deployment.GetDeployment(ws, deploymentID, deploymentName, false, nil, platformCoreClient, astroCoreClient)
        if err != nil {
                return err
        }

        airflowURL, err := getAirflowURL(&requestedDeployment)
        if err != nil {
                return err
        }

        // check key
        if key == "" {
                return errors.New("a variable key is needed to create an airflow variable. Please use the '--key' flag to specify a key")
        }

        return deployment.VariableUpdate(airflowURL, varValue, key, description, airflowAPIClient, out)
}

//nolint:dupl
func deploymentAirflowVariableCopy(cmd *cobra.Command, out io.Writer) error {
        ws, err := coalesceWorkspace()
        if err != nil {
                return errors.Wrap(err, "failed to find a valid workspace")
        }

        // Silence Usage as we have now validated command input
        cmd.SilenceUsage = true

        // get source deployment
        if fromDeploymentName == "" && fromDeploymentID == "" {
                fmt.Println("Which deployment should airflow variables be copied from?")
        }
        fromDeployment, err := deployment.GetDeployment(ws, fromDeploymentID, fromDeploymentName, false, nil, platformCoreClient, astroCoreClient)
        if err != nil {
                return errors.Wrap(err, "failed to find the source Deployment")
        }

        fromAirflowURL, err := getAirflowURL(&fromDeployment)
        if err != nil {
                return errors.Wrap(err, "failed to find the source deployments airflow webserver URL")
        }

        // get target deployment
        if toDeploymentName == "" && toDeploymentID == "" {
                fmt.Println("Which deployment should airflow variables be pasted to?")
        }
        toDeployment, err := deployment.GetDeployment(ws, toDeploymentID, toDeploymentName, false, nil, platformCoreClient, astroCoreClient)
        if err != nil {
                return errors.Wrap(err, "failed to find the target Deployment")
        }

        toAirflowURL, err := getAirflowURL(&toDeployment)
        if err != nil {
                return errors.Wrap(err, "failed to find the target deployments airflow webserver URL")
        }

        fmt.Println(warningVariableCopyCMD)
        return deployment.CopyVariable(fromAirflowURL, toAirflowURL, airflowAPIClient, out)
}

//nolint:dupl
func deploymentPoolList(cmd *cobra.Command, out io.Writer) error {
        ws, err := coalesceWorkspace()
        if err != nil {
                return errors.Wrap(err, "failed to find a valid workspace")
        }

        // Silence Usage as we have now validated command input
        cmd.SilenceUsage = true

        // get or select the deployment
        requestedDeployment, err := deployment.GetDeployment(ws, deploymentID, deploymentName, false, nil, platformCoreClient, astroCoreClient)
        if err != nil {
                return err
        }

        airflowURL, err := getAirflowURL(&requestedDeployment)
        if err != nil {
                return err
        }

        return deployment.PoolList(airflowURL, airflowAPIClient, out)
}

func deploymentPoolCreate(cmd *cobra.Command, out io.Writer) error { //nolint
        ws, err := coalesceWorkspace()
        if err != nil {
                return errors.Wrap(err, "failed to find a valid workspace")
        }

        // Silence Usage as we have now validated command input
        cmd.SilenceUsage = true

        // get or select the deployment
        requestedDeployment, err := deployment.GetDeployment(ws, deploymentID, deploymentName, false, nil, platformCoreClient, astroCoreClient)
        if err != nil {
                return err
        }

        airflowURL, err := getAirflowURL(&requestedDeployment)
        if err != nil {
                return err
        }

        // check name
        if name == "" {
                return errors.New("a pool name is needed to create a pool. Please use the '--name' flag to specify a name")
        }

        // check includeDeferred
        var includeDeferredValue bool
        switch includeDeferred {
        case enable:
                includeDeferredValue = true
        case disable, "":
                includeDeferredValue = false
        default:
                return errors.New("Invalid --include-deferred value")
        }

        return deployment.PoolCreate(airflowURL, name, description, slots, includeDeferredValue, airflowAPIClient, out)
}

func deploymentPoolUpdate(cmd *cobra.Command, out io.Writer) error { //nolint
        ws, err := coalesceWorkspace()
        if err != nil {
                return errors.Wrap(err, "failed to find a valid workspace")
        }

        // Silence Usage as we have now validated command input
        cmd.SilenceUsage = true

        // get or select the deployment
        requestedDeployment, err := deployment.GetDeployment(ws, deploymentID, deploymentName, false, nil, platformCoreClient, astroCoreClient)
        if err != nil {
                return err
        }

        airflowURL, err := getAirflowURL(&requestedDeployment)
        if err != nil {
                return err
        }

        // check name
        if name == "" {
                return errors.New("a pool name is needed to update a pool. Please use the '--name' flag to specify a name")
        }

        // check includeDeferred
        if airflowversions.AirflowMajorVersionForRuntimeVersion(requestedDeployment.RuntimeVersion) >= "3" && includeDeferred == "" {
                return errors.New("an include deferred value is needed to update a pool. Please use the '--include-deferred' flag to specify a value")
        }
        var includeDeferredValue bool
        switch includeDeferred {
        case enable:
                includeDeferredValue = true
        case disable, "":
                includeDeferredValue = false
        default:
                return errors.New("Invalid --include-deferred value")
        }

        return deployment.PoolUpdate(airflowURL, name, description, slots, includeDeferredValue, airflowAPIClient, out)
}

//nolint:dupl
func deploymentPoolCopy(cmd *cobra.Command, out io.Writer) error {
        ws, err := coalesceWorkspace()
        if err != nil {
                return errors.Wrap(err, "failed to find a valid workspace")
        }

        // Silence Usage as we have now validated command input
        cmd.SilenceUsage = true

        // get source deployment
        if fromDeploymentName == "" && fromDeploymentID == "" {
                fmt.Println("Which deployment should pools be copied from?")
        }
        fromDeployment, err := deployment.GetDeployment(ws, fromDeploymentID, fromDeploymentName, false, nil, platformCoreClient, astroCoreClient)
        if err != nil {
                return errors.Wrap(err, "failed to find the source Deployment")
        }

        fromAirflowURL, err := getAirflowURL(&fromDeployment)
        if err != nil {
                return errors.Wrap(err, "failed to find the source deployments airflow webserver URL")
        }

        // get target deployment
        if toDeploymentName == "" && toDeploymentID == "" {
                fmt.Println("Which deployment should pools be pasted to?")
        }
        toDeployment, err := deployment.GetDeployment(ws, toDeploymentID, toDeploymentName, false, nil, platformCoreClient, astroCoreClient)
        if err != nil {
                return errors.Wrap(err, "failed to find the target Deployment")
        }

        toAirflowURL, err := getAirflowURL(&toDeployment)
        if err != nil {
                return errors.Wrap(err, "failed to find the target deployments airflow webserver URL")
        }

        return deployment.CopyPool(fromAirflowURL, toAirflowURL, airflowAPIClient, out)
}

func getAirflowURL(depl *astroplatformcore.Deployment) (string, error) {
        value, err := inspect.ReturnSpecifiedValue(depl, webserverURLField, platformCoreClient)
        if err != nil {
                return "", errors.Wrap(err, "failed to find a deployments airflow webserver URL")
        }

        airflowURL := fmt.Sprintf("%v", value)
        splitAirflowURL := strings.Split(airflowURL, "?")[0]

        return splitAirflowURL, nil
}

astronomer / astro-cli / 357629b1-60cb-43d6-9ac2-8b5249dd574b

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous