fix(deployments datasource): get by name (#343)

* Deployment: Replace List with GetByName

For the Deployment datasource, we were getting it by name by using the
/filter endpoint. But we weren't passing any filter body in the request,
so it was always getting all Deployments back.

This change uses the API endpoint to get a deployment by name, which now
requires a flow name as well.

Related to #330

* Validate either id or name and flow_name

Validates that the correct combination of fields is provided to retrieve
the datasource correctly.

* Specify flow_name in datasource test

* Generate Terraform Docs

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

docs/data-sources/deployment.md

@@ -30,12 +30,14 @@ data "prefect_deployment" "existing_by_id_string" {
 
 # Get deployment by name using Terraform name reference.
 data "prefect_deployment" "existing_by_id_string" {
-  name = prefect_deployment.my_existing_deployment.name
+  flow_name = prefect_flow.my_existing_flow.name
+  name      = prefect_deployment.my_existing_deployment.name
 }
 
 # Get deployment by name string.
 data "prefect_deployment" "existing_by_id_string" {
-  name = "my_existing_deployment"
+  name      = "my_existing_deployment"
+  flow_name = "example_flow"
 }
 ```
 
@@ -45,6 +47,7 @@ data "prefect_deployment" "existing_by_id_string" {
 ### Optional
 
 - `account_id` (String) Account ID (UUID), defaults to the account set in the provider
+- `flow_name` (String) Flow name associated with the deployment
 - `id` (String) Deployment ID (UUID)
 - `name` (String) Name of the deployment
 - `workspace_id` (String) Workspace ID (UUID) to associate deployment to

examples/data-sources/prefect_deployment/data-source.tf

@@ -10,10 +10,12 @@ data "prefect_deployment" "existing_by_id_string" {
 
 # Get deployment by name using Terraform name reference.
 data "prefect_deployment" "existing_by_id_string" {
-  name = prefect_deployment.my_existing_deployment.name
+  flow_name = prefect_flow.my_existing_flow.name
+  name      = prefect_deployment.my_existing_deployment.name
 }
 
 # Get deployment by name string.
 data "prefect_deployment" "existing_by_id_string" {
-  name = "my_existing_deployment"
+  name      = "my_existing_deployment"
+  flow_name = "example_flow"
 }

internal/api/deployments.go

@@ -10,7 +10,7 @@ import (
 type DeploymentsClient interface {
 	Create(ctx context.Context, data DeploymentCreate) (*Deployment, error)
 	Get(ctx context.Context, deploymentID uuid.UUID) (*Deployment, error)
-	List(ctx context.Context, handleNames []string) ([]*Deployment, error)
+	GetByName(ctx context.Context, flowName, deploymentName string) (*Deployment, error)
 	Update(ctx context.Context, deploymentID uuid.UUID, data DeploymentUpdate) error
 	Delete(ctx context.Context, deploymentID uuid.UUID) error
 }

internal/client/deployments.go

@@ -60,29 +60,30 @@ func (c *DeploymentsClient) Create(ctx context.Context, data api.DeploymentCreat
 	return &deployment, nil
 }
 
-// List returns a list of Deployments based on the provided list of names.
-func (c *DeploymentsClient) List(ctx context.Context, _ []string) ([]*api.Deployment, error) {
+// Get returns details for a Deployment by ID.
+func (c *DeploymentsClient) Get(ctx context.Context, deploymentID uuid.UUID) (*api.Deployment, error) {
 	cfg := requestConfig{
-		method:       http.MethodPost,
-		url:          fmt.Sprintf("%s/filter", c.routePrefix),
+		method:       http.MethodGet,
+		url:          fmt.Sprintf("%s/%s", c.routePrefix, deploymentID.String()),
 		body:         http.NoBody,
 		apiKey:       c.apiKey,
 		successCodes: successCodesStatusOK,
 	}
 
-	var deployments []*api.Deployment
-	if err := requestWithDecodeResponse(ctx, c.hc, cfg, &deployments); err != nil {
-		return nil, fmt.Errorf("failed to list deployments: %w", err)
+	var deployment api.Deployment
+	if err := requestWithDecodeResponse(ctx, c.hc, cfg, &deployment); err != nil {
+		return nil, fmt.Errorf("failed to get deployment: %w", err)
 	}
 
-	return deployments, nil
+	return &deployment, nil
 }
 
-// Get returns details for a Deployment by ID.
-func (c *DeploymentsClient) Get(ctx context.Context, deploymentID uuid.UUID) (*api.Deployment, error) {
+// GetByName returns details for a Deployment by name.
+func (c *DeploymentsClient) GetByName(ctx context.Context, flowName, deploymentName string) (*api.Deployment, error) {
+	url := fmt.Sprintf("%s/name/%s/%s", c.routePrefix, flowName, deploymentName)
 	cfg := requestConfig{
 		method:       http.MethodGet,
-		url:          fmt.Sprintf("%s/%s", c.routePrefix, deploymentID.String()),
+		url:          url,
 		body:         http.NoBody,
 		apiKey:       c.apiKey,
 		successCodes: successCodesStatusOK,

internal/provider/datasources/deployment.go

@@ -33,6 +33,9 @@ type deploymentDataSource struct {
 type DeploymentDataSourceModel struct {
 	// The model requires the same fields, so reuse the fields defined in the resource model.
 	resources.DeploymentResourceModel
+
+	// The following fields are specific to the Deployment datasource.
+	FlowName types.String `tfsdk:"flow_name"`
 }
 
 // NewDeploymentDataSource is a helper function to simplify the provider implementation.
@@ -94,6 +97,13 @@ The Deployment ID takes precedence over deployment name.
 				CustomType:  customtypes.UUIDType{},
 				Description: "Flow ID (UUID) to associate deployment to",
 			},
+			// flow_name is specific to the datasource because it's used in the API endpoint
+			// to find a deployment by name.
+			"flow_name": schema.StringAttribute{
+				Computed:    true,
+				Optional:    true,
+				Description: "Flow name associated with the deployment",
+			},
 			"paused": schema.BoolAttribute{
 				Computed:    true,
 				Description: "Whether or not the deployment is paused.",
@@ -243,7 +253,10 @@ func (d *deploymentDataSource) Read(ctx context.Context, req datasource.ReadRequ
 	// If both are set, we prefer the ID
 	var deployment *api.Deployment
 	var operation string
-	if !model.ID.IsNull() {
+	var getErr error
+
+	switch {
+	case !model.ID.IsNull():
 		var deploymentID uuid.UUID
 		deploymentID, err = uuid.Parse(model.ID.ValueString())
 		if err != nil {
@@ -257,25 +270,20 @@ func (d *deploymentDataSource) Read(ctx context.Context, req datasource.ReadRequ
 		}
 
 		operation = "get"
-		deployment, err = client.Get(ctx, deploymentID)
-	} else if !model.Name.IsNull() {
-		var deployments []*api.Deployment
-		operation = "list"
-		deployments, err = client.List(ctx, []string{model.Name.ValueString()})
-
-		if len(deployments) != 1 {
-			resp.Diagnostics.AddError(
-				"Could not find Deployment",
-				fmt.Sprintf("Could not find Deployment with name %s", model.Name.ValueString()),
-			)
+		deployment, getErr = client.Get(ctx, deploymentID)
+	case !model.FlowName.IsNull() && !model.Name.IsNull():
+		operation = "get by name"
+		deployment, getErr = client.GetByName(ctx, model.FlowName.ValueString(), model.Name.ValueString())
+	default:
+		resp.Diagnostics.AddError(
+			"Either id, or name and flow_name are unset",
+			"Please configure either id, or name and flow_name.",
+		)
 
-			return
-		}
-
-		deployment = deployments[0]
+		return
 	}
 
-	if err != nil {
+	if getErr != nil {
 		resp.Diagnostics.Append(helpers.ResourceClientErrorDiagnostic("Deployment", operation, err))
 
 		return

internal/provider/datasources/deployment_test.go

@@ -44,6 +44,7 @@ data "prefect_deployment" "test_by_id" {
 
 data "prefect_deployment" "test_by_name" {
   name = prefect_deployment.test.name
+  flow_name = prefect_flow.test.name
 
   account_id = "%s"
   workspace_id = prefect_workspace.test.id