awss3

package
v1.86.1 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 23, 2026 License: BSD-3-Clause Imports: 28 Imported by: 0

README

AWS S3 Adapter

Entity External IDs

The Entity External ID maps to a CSV filename in an S3 bucket. The adapter constructs the S3 object key as:

{prefix}/{entityExternalID}.{fileType}

The adapter is open-ended - any filename (without extension) can be used as an entity external ID.

Entity External ID S3 Object Key (with prefix data/internal) Description
users data/internal/users.csv Users CSV file
customers data/internal/customers.csv Customers CSV file
sales data/internal/sales.csv Sales CSV file

How It Works

The Entity External ID is combined with the configured path prefix and file type to form the S3 object key. The adapter then fetches that object from the configured S3 bucket using the AWS SDK's GetObject API.

Example: with config bucket: "my-bucket", prefix: "exports/2024", entity external ID "orders", file type "csv":

  • S3 path: s3://my-bucket/exports/2024/orders.csv

Configuration

{
  "region": "us-east-1",
  "bucket": "my-data-bucket",
  "prefix": "data/internal",
  "fileType": "csv"
}

Notes

  • The only supported file type is csv.
  • Max page size is 1000.
  • Attribute external IDs correspond to CSV column headers.
  • Authentication uses AWS access key/secret key credentials.

Adding New Entity External IDs

No code changes required. The adapter uses the entity external ID as the filename (without extension) in the S3 bucket. Simply configure a new entity in SGNL with a name matching the CSV file (e.g., if the file is departments.csv, set the entity external ID to departments).

Ensure the CSV file exists at {prefix}/{entityExternalID}.csv in the configured S3 bucket. Attribute external IDs must match the CSV column headers exactly.

Documentation

Index

Constants

View Source
const FileTypeCSV = "csv"

Variables

View Source
var (
	DefaultFileType    = FileTypeCSV
	SupportedFileTypes = map[string]struct{}{FileTypeCSV: {}}
)
View Source
var (
	UTF8BOM    = []byte{0xEF, 0xBB, 0xBF}
	UTF16LEBOM = []byte{0xFF, 0xFE}
	UTF16BEBOM = []byte{0xFE, 0xFF}
	UTF32LEBOM = []byte{0xFF, 0xFE, 0x00, 0x00}
	UTF32BEBOM = []byte{0x00, 0x00, 0xFE, 0xFF}
)

BOM (Byte Order Mark) patterns for different encodings.

View Source
var ErrEmptyOrMissing = errors.New("empty or missing")

Functions

func CSVHeaders added in v1.44.0

func CSVHeaders(reader *bufio.Reader, maxRowSizeBytes int64) (headers []string, bytesReadForHeader int64, err error)

func GetObjectKeyFromRequest

func GetObjectKeyFromRequest(request *Request) string

func MarshalS3Cursor added in v1.67.0

func MarshalS3Cursor(cursor *S3Cursor) (string, *framework.Error)

MarshalS3Cursor marshals the cursor into a base64 encoded JSON string.

func NewAdapter

func NewAdapter(client Client) framework.Adapter[Config]

NewAdapter instantiates a new Adapter.

func StreamingCSVToPage added in v1.44.0

func StreamingCSVToPage(
	streamReader *bufio.Reader,
	headers []string,
	pageSize int64,
	attrConfig []*framework.AttributeConfig,
	maxProcessingBytesTotal int64,
	maxRowSizeBytes int64,
) (objects []map[string]any, bytesReadFromDataStream int64, hasNext bool, err error)

Types

type Adapter

type Adapter struct {
	Client Client
}

Adapter implements the framework.Adapter interface to query pages of objects from datasources.

func (*Adapter) GetPage

func (a *Adapter) GetPage(ctx context.Context, request *framework.Request[Config]) framework.Response

GetPage is called by SGNL's ingestion service to query a page of objects from a datasource.

func (*Adapter) RequestPageFromDatasource

func (a *Adapter) RequestPageFromDatasource(
	ctx context.Context, request *framework.Request[Config],
) framework.Response

RequestPageFromDatasource requests a page of objects from a datasource.

func (*Adapter) ValidateGetPageRequest

func (a *Adapter) ValidateGetPageRequest(ctx context.Context, request *framework.Request[Config]) *framework.Error

ValidateGetPageRequest validates the fields of the GetPage Request.

type Auth

type Auth struct {
	// AccessKey is the access key to authenticate with the AWS.
	AccessKey string

	// SecretKey is the secret key to authenticate with the AWS.
	SecretKey string

	// Region is the AWS region to query.
	Region string
}

type Client

type Client interface {
	GetPage(ctx context.Context, request *Request) (*Response, *framework.Error)
}

Client is a client that allows querying the datasource which contains JSON objects.

func NewClient

func NewClient(client *http.Client, awsConfig *aws.Config, maxRowSizeBytes, maxPageSizeBytes int64) (Client, error)

NewClient returns a Client to query the datasource.

type Config

type Config struct {
	// Common configuration
	*config.CommonConfig

	// Region is the AWS region to query.
	Region string `json:"region"`

	// Bucket is the AWS S3 bucket containing the files with entity data.
	Bucket string `json:"bucket"`

	// Prefix is the prefix of the path containing the files with entity data.
	Prefix string `json:"prefix"`

	// FileType is the extension of the files containing the entity data.
	// This defaults to "csv".
	FileType *string `json:"fileType,omitempty"`
}

func (*Config) Validate

func (c *Config) Validate(_ context.Context) error

ValidateConfig validates that a Config received in a GetPage call is valid.

type Datasource

type Datasource struct {
	Client                   *http.Client
	AWSConfig                *aws.Config
	MaxCSVRowSizeBytes       int64
	MaxBytesToProcessPerPage int64
}

func (*Datasource) GetPage

func (d *Datasource) GetPage(ctx context.Context, request *Request) (*Response, *framework.Error)

type Request

type Request struct {
	Auth

	// Bucket is the AWS S3 bucket containing the files with entity data.
	Bucket string

	// PathPrefix is the prefix of the path containing the files with entity data.
	PathPrefix string

	// FileType is the extension of the files containing the entity data.
	FileType string

	// PageSize is the maximum number of objects to return from the entity.
	PageSize int64

	// EntityExternalID is the external ID of the entity.
	// The external ID should match the file name.
	EntityExternalID string

	// Cursor identifies the first object of the page to return, as returned by
	// the last request for the entity.
	// nil in the request for the first page.
	Cursor *S3Cursor

	// RequestTimeoutSeconds is the timeout duration for requests made to datasources.
	// This should be set to the number of seconds to wait before timing out.
	RequestTimeoutSeconds int

	// Ordered is a boolean that indicates whether the results should be ordered.
	Ordered bool

	// AttributeConfig is the list of attributes requested by the datasource.
	AttributeConfig []*framework.AttributeConfig
}

Request is a request to the datasource.

type Response

type Response struct {
	// StatusCode is an HTTP status code.
	StatusCode int

	// RetryAfterHeader is the Retry-After response HTTP header, if set.
	RetryAfterHeader string

	// Objects is the list of items returned by the datasource.
	// May be empty.
	Objects []map[string]any

	// NextCursor is the cursor that identifies the first object of the next page.
	// nil if this is the last page in this full sync.
	NextCursor *S3Cursor
}

Response is a response returned by the datasource.

type S3Cursor added in v1.67.0

type S3Cursor struct {
	// Cursor is the byte position offset in the file where the next S3 fetch should start.
	Cursor *int64 `json:"cursor,omitempty"`

	// Headers contains the parsed CSV headers from the first page.
	// Cached to avoid re-fetching headers on subsequent pages.
	Headers []string `json:"headers,omitempty"`

	// Remainder contains unprocessed bytes from the previous fetch.
	// These bytes are prepended to the next S3 fetch to avoid data loss.
	Remainder []byte `json:"remainder,omitempty"`
}

S3Cursor contains pagination state for S3 CSV files. Headers are cached to avoid re-fetching on subsequent pages.

func UnmarshalS3Cursor added in v1.67.0

func UnmarshalS3Cursor(cursor string) (*S3Cursor, *framework.Error)

UnmarshalS3Cursor unmarshals the cursor from a base64 encoded JSON string. Returns nil cursor if the input is empty.

func (*S3Cursor) MarshalLogObject added in v1.67.0

func (c *S3Cursor) MarshalLogObject(enc zapcore.ObjectEncoder) error

MarshalLogObject implements zapcore.ObjectMarshaler to control cursor logging. Logs metadata (byte position, headers count, remainder length) without exposing actual data.

type S3Handler

type S3Handler struct {
	Client *s3.Client
}

func (*S3Handler) FileExists

func (s *S3Handler) FileExists(ctx context.Context, bucket string, key string) (*s3.HeadObjectOutput, error)

func (*S3Handler) GetFileSize added in v1.44.0

func (s *S3Handler) GetFileSize(ctx context.Context, bucket string, key string) (int64, error)

func (*S3Handler) GetObjectStream added in v1.44.0

func (s *S3Handler) GetObjectStream(ctx context.Context, bucket string, key string, rangeHeader *string) (
	*s3.GetObjectOutput, error)

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL