handler

Documentation

Index

• type CantabularClient
• type DatasetAPIClient
• type Error
  • func NewError(err error, logData map[string]interface{}) *Error
  • func (e *Error) Error() string
  • func (e *Error) LogData() map[string]interface{}
  • func (e *Error) Unwrap() error
• type Generator
• type InstanceComplete
  • func NewInstanceComplete(cfg config.Config, c CantabularClient, d DatasetAPIClient, ...) *InstanceComplete
  • func (h *InstanceComplete) GetS3ContentLength(ctx context.Context, instanceID string, isPublished bool) (int, error)
  • func (h *InstanceComplete) Handle(ctx context.Context, workerID int, msg kafka.Message) error
  • func (h *InstanceComplete) ProduceExportCompleteEvent(instanceID string, isPublished bool, s3Url string, rowCount int32) error
  • func (h *InstanceComplete) UpdateInstance(ctx context.Context, instanceID string, size int, isPublished bool, ...) error
  • func (h *InstanceComplete) UploadCSVFile(ctx context.Context, instanceID string, isPublished bool, ...) (string, int32, error)
  • func (h *InstanceComplete) ValidateInstance(i dataset.Instance) (bool, error)
• type S3Uploader
• type VaultClient

Types

type CantabularClient

type CantabularClient interface {
	StaticDatasetQueryStreamCSV(ctx context.Context, req cantabular.StaticDatasetQueryRequest, consume cantabular.Consumer) (rowCount int32, err error)
}

CantabularClient contains the required method for the Cantabular Client

type DatasetAPIClient

type DatasetAPIClient interface {
	GetInstance(ctx context.Context, userAuthToken, serviceAuthToken, collectionID, instanceID, ifMatch string) (i dataset.Instance, eTag string, err error)
	PutInstance(ctx context.Context, userAuthToken, serviceAuthToken, collectionID, instanceID string, instanceUpdate dataset.UpdateInstance, ifMatch string) (eTag string, err error)
}

DatasetAPIClient contains the required method for the Dataset API Client

type Error

type Error struct {
	// contains filtered or unexported fields
}

Error is the handler package's error type. Is not meant to be compared as a type, but information should be extracted via the interfaces it implements with callback functions. Is not guaranteed to remain exported so shouldn't be treated as such.

func NewError

func NewError(err error, logData map[string]interface{}) *Error

NewError creates a new Error

func (*Error) Error

func (e *Error) Error() string

Error implements the Go standard error interface

func (*Error) LogData

func (e *Error) LogData() map[string]interface{}

LogData implements the DataLogger interface which allows you extract embedded log.Data from an error

func (*Error) Unwrap

func (e *Error) Unwrap() error

type Generator

type Generator interface {
	NewPSK() ([]byte, error)
}

Generator contains methods for dynamically required strings and tokens e.g. UUIDs, PSKs.

type InstanceComplete

type InstanceComplete struct {
	// contains filtered or unexported fields
}

InstanceComplete is the handle for the InstanceCompleteHandler event

func NewInstanceComplete

func NewInstanceComplete(cfg config.Config, c CantabularClient, d DatasetAPIClient, sPrivate, sPublic S3Uploader, v VaultClient, p kafka.IProducer, g Generator) *InstanceComplete

NewInstanceComplete creates a new InstanceCompleteHandler

func (*InstanceComplete) GetS3ContentLength

func (h *InstanceComplete) GetS3ContentLength(ctx context.Context, instanceID string, isPublished bool) (int, error)

GetS3ContentLength obtains an S3 file size (in number of bytes) by calling Head Object

func (*InstanceComplete) Handle

func (h *InstanceComplete) Handle(ctx context.Context, workerID int, msg kafka.Message) error

Handle takes a single event.

func (*InstanceComplete) ProduceExportCompleteEvent

func (h *InstanceComplete) ProduceExportCompleteEvent(instanceID string, isPublished bool, s3Url string, rowCount int32) error

ProduceExportCompleteEvent sends the final kafka message signifying the export complete

func (*InstanceComplete) UpdateInstance

func (h *InstanceComplete) UpdateInstance(ctx context.Context, instanceID string, size int, isPublished bool, s3Url string) error

UpdateInstance updates the instance downlad CSV link using dataset API PUT /instances/{id} endpoint if the instance is published, then the s3Url will be set as public link and the instance state will be set to published otherwise, a private url will be generated and the state will not be changed

func (*InstanceComplete) UploadCSVFile

func (h *InstanceComplete) UploadCSVFile(ctx context.Context, instanceID string, isPublished bool, req cantabular.StaticDatasetQueryRequest) (string, int32, error)

UploadCSVFile queries a static dataset to cantabular using the provided request, transforms the response to a CSV format and streams the data to S3. If the data is published, the S3 file will be stored in the public bucket If the data is private, the S3 file will be stored in the private bucket (encrypted or un-encrypted depending on EncryptionDisabled flag) Returns S3 file URL, file size bytes, number of rows, and any error that happens.

func (*InstanceComplete) ValidateInstance

func (h *InstanceComplete) ValidateInstance(i dataset.Instance) (bool, error)

ValidateInstance validates the instance returned from dp-dataset-api Returns isPublished bool value and any validation error

type S3Uploader

type S3Uploader interface {
	Head(key string) (*s3.HeadObjectOutput, error)
	UploadWithContext(ctx context.Context, input *s3manager.UploadInput, options ...func(*s3manager.Uploader)) (*s3manager.UploadOutput, error)
	UploadWithPSK(input *s3manager.UploadInput, psk []byte) (*s3manager.UploadOutput, error)
	BucketName() string
}

S3Uploader contains the required method for the S3 Uploader

type VaultClient

type VaultClient interface {
	WriteKey(path, key, value string) error
}

VaultClient contains the required methods for the Vault Client