Merge pull request #14 from incident-io/rob/add-incident-updates-stream

Add incident update stream

Author: rliddler

model/incident_update_v2.go

@@ -0,0 +1,45 @@
+package model
+
+import "github.com/incident-io/singer-tap/client"
+
+type incidentUpdateV2 struct{}
+
+var IncidentUpdateV2 incidentUpdateV2
+
+func (incidentUpdateV2) Schema() Property {
+	return Property{
+		Types: []string{"object"},
+		Properties: map[string]Property{
+			"id": {
+				Types: []string{"string"},
+			},
+			"incident_id": {
+				Types: []string{"string"},
+			},
+			"message": {
+				Types: []string{"string", "null"},
+			},
+			"new_incident_status": IncidentStatusV1.Schema(),
+			"new_severity":        Optional(SeverityV2.Schema()),
+			"updater":             ActorV2.Schema(),
+			"created_at":          DateTime.Schema(),
+		},
+	}
+}
+
+func (incidentUpdateV2) Serialize(input client.IncidentUpdateV2) map[string]any {
+	var severity map[string]any
+	if input.NewSeverity != nil {
+		severity = SeverityV2.Serialize(input.NewSeverity)
+	}
+
+	return map[string]any{
+		"id":                  input.Id,
+		"incident_id":         input.IncidentId,
+		"message":             input.Message,
+		"new_incident_status": IncidentStatusV1.Serialize(input.NewIncidentStatus),
+		"new_severity":        severity,
+		"updater":             ActorV2.Serialize(input.Updater),
+		"created_at":          input.CreatedAt,
+	}
+}

tap/stream_incident_updates.go

@@ -0,0 +1,60 @@
+package tap
+
+import (
+	"context"
+
+	kitlog "github.com/go-kit/log"
+	"github.com/incident-io/singer-tap/client"
+	"github.com/incident-io/singer-tap/model"
+	"github.com/pkg/errors"
+	"github.com/samber/lo"
+)
+
+func init() {
+	register(&StreamIncidentUpdates{})
+}
+
+type StreamIncidentUpdates struct {
+}
+
+func (s *StreamIncidentUpdates) Output() *Output {
+	return &Output{
+		Type:   OutputTypeSchema,
+		Stream: "incident_updates",
+		Schema: &model.Schema{
+			HasAdditionalProperties: false,
+			Type:                    []string{"object"},
+			Properties:              model.IncidentUpdateV2.Schema().Properties,
+		},
+		KeyProperties:      []string{"id"},
+		BookmarkProperties: []string{},
+	}
+}
+
+func (s *StreamIncidentUpdates) GetRecords(ctx context.Context, logger kitlog.Logger, cl *client.ClientWithResponses) ([]map[string]any, error) {
+	var (
+		after    *string
+		pageSize = 250
+		results  = []map[string]any{}
+	)
+
+	for {
+		logger.Log("msg", "loading page", "page_size", pageSize, "after", after)
+		page, err := cl.IncidentUpdatesV2ListWithResponse(ctx, &client.IncidentUpdatesV2ListParams{
+			PageSize: &pageSize,
+			After:    after,
+		})
+		if err != nil {
+			return nil, errors.Wrap(err, "listing incident updates")
+		}
+
+		for _, element := range page.JSON200.IncidentUpdates {
+			results = append(results, model.IncidentUpdateV2.Serialize(element))
+		}
+		if count := len(page.JSON200.IncidentUpdates); count == 0 {
+			return results, nil // end pagination
+		} else {
+			after = lo.ToPtr(page.JSON200.IncidentUpdates[count-1].Id)
+		}
+	}
+}