13829663743

Committed 13 Mar 2025 07:55AM UTC coverage: 90.604% (+0.5%) from 90.121%

Build # 13829663743

Build Type

Pull #156

github

Committed by

thschmitt

Commit Message

Add network package with common http client

Moved common code for handling HTTP requests in a new shared
network package to avoid duplication and ensure that each plugin
handles network requests the same way.

The common HTTP client code takes care of a couple of cross-cutting
concerns needed by all plugins and the main HTTP executor:

- Logs request and response when debug flag is set. Added a new
  resettableReader which preserves the request and response bodies while
  they are being read and forwards them to the debug logger.

- Added retries for all HTTP requests. This also leverages the
  resettableReader to ensure the request body can be replayed.

- The `CommandBuilder` generates now an operation id which is set on every
  request the uipathcli performs. The same operation id is kept for the
  duration of the whole command execution which makes it easier to
  correlate multiple requests performed by a single command.

- The `HttpClient` also sets transport-related settings like certificate
  validation and response header timeout.

- Using the built-in context instead of a custom requestError channel.

Pull Request Pull Request #156: Add network package with common http client

Run Details

478 of 505 new or added lines in 26 files covered. (94.65%)

8 existing lines in 5 files now uncovered.

5371 of 5928 relevant lines covered (90.6%)

1.02 hits per line

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

90.51

/plugin/orchestrator/upload_command.go

package orchestrator

import (
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "io"
        "net/http"
        "net/url"
        "strings"

        "github.com/UiPath/uipathcli/log"
        "github.com/UiPath/uipathcli/output"
        "github.com/UiPath/uipathcli/plugin"
        "github.com/UiPath/uipathcli/utils/network"
        "github.com/UiPath/uipathcli/utils/stream"
        "github.com/UiPath/uipathcli/utils/visualization"
)

// The UploadCommand is a custom command for the orchestrator service which makes uploading
// files more convenient. It provides a wrapper over retrieving the write url and actually
// performing the upload.
type UploadCommand struct{}

func (c UploadCommand) Command() plugin.Command {
        return *plugin.NewCommand("orchestrator").
                WithCategory("buckets", "Orchestrator Buckets", "Buckets provide a per-folder storage solution for RPA developers to leverage in creating automation projects.").
                WithOperation("upload", "Upload file", "Uploads the provided file to the bucket").
                WithParameter("folder-id", plugin.ParameterTypeInteger, "Folder/OrganizationUnit Id", true).
                WithParameter("key", plugin.ParameterTypeInteger, "The Bucket Id", true).
                WithParameter("path", plugin.ParameterTypeString, "The BlobFile full path", true).
                WithParameter("file", plugin.ParameterTypeBinary, "The file to upload", true)
}

func (c UploadCommand) Execute(ctx plugin.ExecutionContext, writer output.OutputWriter, logger log.Logger) error {
        writeUrl, err := c.getWriteUrl(ctx, logger)
        if err != nil {
                return err
        }
        return c.upload(ctx, logger, writeUrl)
}

func (c UploadCommand) upload(ctx plugin.ExecutionContext, logger log.Logger, url string) error {
        uploadBar := visualization.NewProgressBar(logger)
        defer uploadBar.Remove()
        context, cancel := context.WithCancelCause(context.Background())
        request := c.createUploadRequest(ctx, url, uploadBar, cancel)
        client := network.NewHttpClient(logger, c.httpClientSettings(ctx))
        response, err := client.SendWithContext(request, context)
        if err != nil {
                return err
        }
        defer response.Body.Close()
        body, err := io.ReadAll(response.Body)
        if err != nil {
                return fmt.Errorf("Error reading response: %w", err)
        }
        if response.StatusCode != http.StatusCreated {
                return fmt.Errorf("Orchestrator returned status code '%v' and body '%v'", response.StatusCode, string(body))
        }
        return nil
}

func (c UploadCommand) createUploadRequest(ctx plugin.ExecutionContext, url string, uploadBar *visualization.ProgressBar, cancel context.CancelCauseFunc) *network.HttpRequest {
        file := ctx.Input
        if file == nil {
                file = c.getFileParameter(ctx.Parameters)
        }
        bodyReader, bodyWriter := io.Pipe()
        contentType, contentLength := c.writeBody(bodyWriter, file, cancel)
        uploadReader := c.progressReader("uploading...", "completing  ", bodyReader, contentLength, uploadBar)

        header := http.Header{
                "Content-Type":   {contentType},
                "x-ms-blob-type": {"BlockBlob"},
        }
        return network.NewHttpPutRequest(url, header, uploadReader, contentLength)
}

func (c UploadCommand) writeBody(bodyWriter *io.PipeWriter, input stream.Stream, cancel context.CancelCauseFunc) (string, int64) {
        go func() {
                defer bodyWriter.Close()
                data, err := input.Data()
                if err != nil {
                        cancel(err)
                        return
                }
                defer data.Close()
                _, err = io.Copy(bodyWriter, data)
                if err != nil {
                        cancel(err)
                        return
                }
        }()
        size, _ := input.Size()
        return "application/octet-stream", size
}

func (c UploadCommand) progressReader(text string, completedText string, reader io.Reader, length int64, progressBar *visualization.ProgressBar) io.Reader {
        if length < 10*1024*1024 {
                return reader
        }
        return visualization.NewProgressReader(reader, func(progress visualization.Progress) {
                displayText := text
                if progress.Completed {
                        displayText = completedText
                }
                progressBar.UpdateProgress(displayText, progress.BytesRead, length, progress.BytesPerSecond)
        })
}

func (c UploadCommand) getWriteUrl(ctx plugin.ExecutionContext, logger log.Logger) (string, error) {
        if ctx.Organization == "" {
                return "", errors.New("Organization is not set")
        }
        if ctx.Tenant == "" {
                return "", errors.New("Tenant is not set")
        }
        request := c.createWriteUrlRequest(ctx)
        client := network.NewHttpClient(logger, c.httpClientSettings(ctx))
        response, err := client.Send(request)
        if err != nil {
                return "", err
        }
        defer response.Body.Close()
        body, err := io.ReadAll(response.Body)
        if err != nil {
                return "", fmt.Errorf("Error reading response: %w", err)
        }
        if response.StatusCode != http.StatusOK {
                return "", fmt.Errorf("Orchestrator returned status code '%v' and body '%v'", response.StatusCode, string(body))
        }
        var result urlResponse
        err = json.Unmarshal(body, &result)
        if err != nil {
                return "", fmt.Errorf("Error parsing json response: %w", err)
        }
        return result.Uri, nil
}

func (c UploadCommand) createWriteUrlRequest(ctx plugin.ExecutionContext) *network.HttpRequest {
        folderId := c.getIntParameter("folder-id", ctx.Parameters)
        bucketId := c.getIntParameter("key", ctx.Parameters)
        path := c.getStringParameter("path", ctx.Parameters)

        uri := c.formatUri(ctx.BaseUri, ctx.Organization, ctx.Tenant) + fmt.Sprintf("/odata/Buckets(%d)/UiPath.Server.Configuration.OData.GetWriteUri?path=%s", bucketId, path)
        header := http.Header{
                "X-UiPath-OrganizationUnitId": {fmt.Sprintf("%d", folderId)},
        }
        for key, value := range ctx.Auth.Header {
                header.Set(key, value)
        }
        return network.NewHttpGetRequest(uri, header)
}

func (c UploadCommand) formatUri(baseUri url.URL, org string, tenant string) string {
        path := baseUri.Path
        if baseUri.Path == "" {
                path = "/{organization}/{tenant}/orchestrator_"
        }
        path = strings.ReplaceAll(path, "{organization}", org)
        path = strings.ReplaceAll(path, "{tenant}", tenant)
        path = strings.TrimSuffix(path, "/")
        return fmt.Sprintf("%s://%s%s", baseUri.Scheme, baseUri.Host, path)
}

func (c UploadCommand) getStringParameter(name string, parameters []plugin.ExecutionParameter) string {
        result := ""
        for _, p := range parameters {
                if p.Name == name {
                        if data, ok := p.Value.(string); ok {
                                result = data
                                break
                        }
                }
        }
        return result
}

func (c UploadCommand) getIntParameter(name string, parameters []plugin.ExecutionParameter) int {
        result := 0
        for _, p := range parameters {
                if p.Name == name {
                        if data, ok := p.Value.(int); ok {
                                result = data
                                break
                        }
                }
        }
        return result
}

func (c UploadCommand) getFileParameter(parameters []plugin.ExecutionParameter) stream.Stream {
        var result stream.Stream
        for _, p := range parameters {
                if p.Name == "file" {
                        if stream, ok := p.Value.(stream.Stream); ok {
                                result = stream
                                break
                        }
                }
        }
        return result
}

func (c UploadCommand) httpClientSettings(ctx plugin.ExecutionContext) network.HttpClientSettings {
        return *network.NewHttpClientSettings(
                ctx.Debug,
                ctx.Settings.OperationId,
                ctx.Settings.Timeout,
                ctx.Settings.MaxAttempts,
                ctx.Settings.Insecure)
}

func NewUploadCommand() *UploadCommand {
        return &UploadCommand{}
}

1	package orchestrator
2
3	import (
4	"context"
5	"encoding/json"
6	"errors"
7	"fmt"
8	"io"
9	"net/http"
10	"net/url"
11	"strings"
12
13	"github.com/UiPath/uipathcli/log"
14	"github.com/UiPath/uipathcli/output"
15	"github.com/UiPath/uipathcli/plugin"
16	"github.com/UiPath/uipathcli/utils/network"
17	"github.com/UiPath/uipathcli/utils/stream"
18	"github.com/UiPath/uipathcli/utils/visualization"
19	)
20
21	// The UploadCommand is a custom command for the orchestrator service which makes uploading
22	// files more convenient. It provides a wrapper over retrieving the write url and actually
23	// performing the upload.
24	type UploadCommand struct{}
25
26	func (c UploadCommand) Command() plugin.Command {	1✔
27	return *plugin.NewCommand("orchestrator").	1✔
28	WithCategory("buckets", "Orchestrator Buckets", "Buckets provide a per-folder storage solution for RPA developers to leverage in creating automation projects.").	1✔
29	WithOperation("upload", "Upload file", "Uploads the provided file to the bucket").	1✔
30	WithParameter("folder-id", plugin.ParameterTypeInteger, "Folder/OrganizationUnit Id", true).	1✔
31	WithParameter("key", plugin.ParameterTypeInteger, "The Bucket Id", true).	1✔
32	WithParameter("path", plugin.ParameterTypeString, "The BlobFile full path", true).	1✔
33	WithParameter("file", plugin.ParameterTypeBinary, "The file to upload", true)	1✔
34	}	1✔
35
36	func (c UploadCommand) Execute(ctx plugin.ExecutionContext, writer output.OutputWriter, logger log.Logger) error {	1✔
37	writeUrl, err := c.getWriteUrl(ctx, logger)	1✔
38	if err != nil {	2✔
39	return err	1✔
40	}	1✔
41	return c.upload(ctx, logger, writeUrl)	1✔
42	}
43
44	func (c UploadCommand) upload(ctx plugin.ExecutionContext, logger log.Logger, url string) error {	1✔
45	uploadBar := visualization.NewProgressBar(logger)	1✔
46	defer uploadBar.Remove()	1✔
47	context, cancel := context.WithCancelCause(context.Background())	1✔
48	request := c.createUploadRequest(ctx, url, uploadBar, cancel)	1✔
49	client := network.NewHttpClient(logger, c.httpClientSettings(ctx))	1✔
50	response, err := client.SendWithContext(request, context)	1✔
51	if err != nil {	2✔
52	return err	1✔
53	}	1✔
54	defer response.Body.Close()	1✔
55	body, err := io.ReadAll(response.Body)	1✔
56	if err != nil {	1✔
57	return fmt.Errorf("Error reading response: %w", err)	×
58	}	×
59	if response.StatusCode != http.StatusCreated {	1✔
UNCOV 60	return fmt.Errorf("Orchestrator returned status code '%v' and body '%v'", response.StatusCode, string(body))	×
61	}	×
62	return nil	1✔
63	}
64
65	func (c UploadCommand) createUploadRequest(ctx plugin.ExecutionContext, url string, uploadBar visualization.ProgressBar, cancel context.CancelCauseFunc) network.HttpRequest {	1✔
66	file := ctx.Input	1✔
67	if file == nil {	2✔
68	file = c.getFileParameter(ctx.Parameters)	1✔
69	}	1✔
70	bodyReader, bodyWriter := io.Pipe()	1✔
71	contentType, contentLength := c.writeBody(bodyWriter, file, cancel)	1✔
72	uploadReader := c.progressReader("uploading...", "completing ", bodyReader, contentLength, uploadBar)	1✔
73		1✔
74	header := http.Header{	1✔
75	"Content-Type": {contentType},	1✔
76	"x-ms-blob-type": {"BlockBlob"},	1✔
77	}	1✔
78	return network.NewHttpPutRequest(url, header, uploadReader, contentLength)	1✔
79	}
80
81	func (c UploadCommand) writeBody(bodyWriter *io.PipeWriter, input stream.Stream, cancel context.CancelCauseFunc) (string, int64) {	1✔
82	go func() {	2✔
83	defer bodyWriter.Close()	1✔
84	data, err := input.Data()	1✔
85	if err != nil {	2✔
86	cancel(err)	1✔
87	return	1✔
88	}	1✔
89	defer data.Close()	1✔
90	_, err = io.Copy(bodyWriter, data)	1✔
91	if err != nil {	1✔
NEW 92	cancel(err)	×
93	return	×
94	}	×
95	}()
96	size, _ := input.Size()	1✔
97	return "application/octet-stream", size	1✔
98	}
99
100	func (c UploadCommand) progressReader(text string, completedText string, reader io.Reader, length int64, progressBar *visualization.ProgressBar) io.Reader {	1✔
101	if length < 1010241024 {	2✔
102	return reader	1✔
103	}	1✔
104	return visualization.NewProgressReader(reader, func(progress visualization.Progress) {	2✔
105	displayText := text	1✔
106	if progress.Completed {	2✔
107	displayText = completedText	1✔
108	}	1✔
109	progressBar.UpdateProgress(displayText, progress.BytesRead, length, progress.BytesPerSecond)	1✔
110	})
111	}
112
113	func (c UploadCommand) getWriteUrl(ctx plugin.ExecutionContext, logger log.Logger) (string, error) {	1✔
114	if ctx.Organization == "" {	2✔
115	return "", errors.New("Organization is not set")	1✔
116	}	1✔
117	if ctx.Tenant == "" {	2✔
118	return "", errors.New("Tenant is not set")	1✔
119	}	1✔
120	request := c.createWriteUrlRequest(ctx)	1✔
121	client := network.NewHttpClient(logger, c.httpClientSettings(ctx))	1✔
122	response, err := client.Send(request)	1✔
123	if err != nil {	1✔
NEW 124	return "", err	×
125	}	×
126	defer response.Body.Close()	1✔
127	body, err := io.ReadAll(response.Body)	1✔
128	if err != nil {	1✔
129	return "", fmt.Errorf("Error reading response: %w", err)	×
130	}	×
131	if response.StatusCode != http.StatusOK {	2✔
132	return "", fmt.Errorf("Orchestrator returned status code '%v' and body '%v'", response.StatusCode, string(body))	1✔
133	}	1✔
134	var result urlResponse	1✔
135	err = json.Unmarshal(body, &result)	1✔
136	if err != nil {	1✔
137	return "", fmt.Errorf("Error parsing json response: %w", err)	×
138	}	×
139	return result.Uri, nil	1✔
140	}
141
142	func (c UploadCommand) createWriteUrlRequest(ctx plugin.ExecutionContext) *network.HttpRequest {	1✔
143	folderId := c.getIntParameter("folder-id", ctx.Parameters)	1✔
144	bucketId := c.getIntParameter("key", ctx.Parameters)	1✔
145	path := c.getStringParameter("path", ctx.Parameters)	1✔
146		1✔
147	uri := c.formatUri(ctx.BaseUri, ctx.Organization, ctx.Tenant) + fmt.Sprintf("/odata/Buckets(%d)/UiPath.Server.Configuration.OData.GetWriteUri?path=%s", bucketId, path)	1✔
148	header := http.Header{	1✔
149	"X-UiPath-OrganizationUnitId": {fmt.Sprintf("%d", folderId)},	1✔
150	}	1✔
151	for key, value := range ctx.Auth.Header {	1✔
NEW 152	header.Set(key, value)	×
UNCOV 153	}	×
154	return network.NewHttpGetRequest(uri, header)	1✔
155	}
156
157	func (c UploadCommand) formatUri(baseUri url.URL, org string, tenant string) string {	1✔
158	path := baseUri.Path	1✔
159	if baseUri.Path == "" {	2✔
160	path = "/{organization}/{tenant}/orchestrator_"	1✔
161	}	1✔
162	path = strings.ReplaceAll(path, "{organization}", org)	1✔
163	path = strings.ReplaceAll(path, "{tenant}", tenant)	1✔
164	path = strings.TrimSuffix(path, "/")	1✔
165	return fmt.Sprintf("%s://%s%s", baseUri.Scheme, baseUri.Host, path)	1✔
166	}
167
168	func (c UploadCommand) getStringParameter(name string, parameters []plugin.ExecutionParameter) string {	1✔
169	result := ""	1✔
170	for _, p := range parameters {	2✔
171	if p.Name == name {	2✔
172	if data, ok := p.Value.(string); ok {	2✔
173	result = data	1✔
174	break	1✔
175	}
176	}
177	}
178	return result	1✔
179	}
180
181	func (c UploadCommand) getIntParameter(name string, parameters []plugin.ExecutionParameter) int {	1✔
182	result := 0	1✔
183	for _, p := range parameters {	2✔
184	if p.Name == name {	2✔
185	if data, ok := p.Value.(int); ok {	2✔
186	result = data	1✔
187	break	1✔
188	}
189	}
190	}
191	return result	1✔
192	}
193
194	func (c UploadCommand) getFileParameter(parameters []plugin.ExecutionParameter) stream.Stream {	1✔
195	var result stream.Stream	1✔
196	for _, p := range parameters {	2✔
197	if p.Name == "file" {	2✔
198	if stream, ok := p.Value.(stream.Stream); ok {	2✔
199	result = stream	1✔
200	break	1✔
201	}
202	}
203	}
204	return result	1✔
205	}
206
207	func (c UploadCommand) httpClientSettings(ctx plugin.ExecutionContext) network.HttpClientSettings {	1✔
208	return *network.NewHttpClientSettings(	1✔
209	ctx.Debug,	1✔
210	ctx.Settings.OperationId,	1✔
211	ctx.Settings.Timeout,	1✔
212	ctx.Settings.MaxAttempts,	1✔
213	ctx.Settings.Insecure)	1✔
214	}	1✔
215
216	func NewUploadCommand() *UploadCommand {	1✔
217	return &UploadCommand{}	1✔
218	}	1✔

UiPath / uipathcli / 13829663743

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