S3 reader

Reads an object from Amazon S3 (or an S3-compatible store such as MinIO) and emits it as a file stream, for example for a CSV or ASN.1 decoder. Every object is a batch: once it is processed and committed downstream, the object can be deleted, moved or tagged; a failed object can be moved to an error location. The object comes from an incoming object file info (S3 enumerator) or from the configuration. If the object does not exist, the input is passed to NoData.

Pins

Pin	Kind	Types	Description
Input	input	Viaducta.TypeSystem.Metadata.ViaFileInfo	Triggers the read. A ViaFileInfo whose FilePath is an s3:// URL selects the object; any other input reads the configured object.
Output	output	Viaducta.Plugins.Files.Types.ViaFileStream	The object stream with name, size and date, sent in the batch of the object.
NoData	no-data output	same as input	Called with the input record when the bucket or object does not exist (HTTP 404).

Configuration

Field	Type	Default	Description
Authentication	dropdown: 0 AccessKey, 1 DefaultChain, 2 AssumeRole	0	How to sign in.
AccessKeyId	string		Access key id. Required for 0. For 2 optional: with it the role is assumed with this key, without it with the default chain.
SecretAccessKey	password		Secret of AccessKeyId. Required with AccessKeyId.
SessionToken	password		Session token of temporary credentials; optional.
RoleArn	string		2: ARN of the role to assume. Required for 2.
ExternalId	string		2: external id the trust policy of the role requires; optional.
StsServiceUrl	string		2: STS endpoint. Empty: the regional AWS STS endpoint of Region.
Region	string		AWS region of the bucket, for example eu-central-1. Required when ServiceUrl is empty; with ServiceUrl the signing region (empty: us-east-1).
ServiceUrl	string		Endpoint of an S3-compatible store or an S3 VPC endpoint, http(s) without query. Empty: AWS S3 in Region.
ForcePathStyle	boolean	false	true addresses <endpoint>/<bucket>/<key>; most S3-compatible stores need it.
BucketName	string		Bucket of the object, used when the input is not an object ViaFileInfo.
Key	string		Object key including folders, for example daily/export.csv, used when the input is not an object ViaFileInfo.
PostProcessing	dropdown: 0 None, 1 Delete, 2 Move, 3 Tag	0	What happens to the object after its batch committed.
PostProcessBucket	string		2 Move: target bucket (must exist); empty keeps the object's bucket.
PostProcessPrefix	string		2 Move: put in front of the key, for example done/. Move needs this or PostProcessBucket.
PostProcessTag	string	viaducta-processed	3 Tag: object tag to set, name or name=value; without a value the UTC time (yyyy-MM-ddTHH:mm:ssZ) is the value.
ErrorBucket	string		Bucket a failed object is moved to (must exist); empty keeps the object's bucket.
ErrorPrefix	string		Put in front of the key of a failed object, for example error/. With ErrorBucket and ErrorPrefix empty, failed objects stay where they are.

Behaviour

• The S3 client is created at flow start; sign-in and post-processing settings are checked then (Move without target, Tag without name fail the start). Sign-in modes as in S3 enumerator. The identity needs s3:GetObject, plus s3:DeleteObject for Delete, s3:PutObject, s3:DeleteObject, s3:GetObjectTagging and s3:PutObjectTagging for Move (and the error location), s3:GetObjectTagging and s3:PutObjectTagging for Tag.
• Per input the object to read is chosen: when the input is a ViaFileInfo whose FilePath is s3://<bucket>/<key> (as emitted by S3 enumerator), bucket and key are taken from it; otherwise BucketName/Key are used. The endpoint always comes from the sign-in. With neither, the input fails with an error.
• A 404 for the GET calls NoData with the input record (no batch); other errors (authentication, network) raise an error.
• Otherwise a batch labelled with the key is opened and Output is called with a ViaFileStream in that batch: Stream (the response body, read forward as it downloads), FilePath = <bucket>/<key>, Name = key, Extension from the key, Size = content length, CreateDateTime and LastAccessTime = last modification (UTC). Stream, size, date and ETag come from the same GET, so they belong to one version of the object. When Output returns the stream is closed and the batch completes: downstream sinks (database writers, ...) commit.
• After the commit the object is post-processed: 1 Delete deletes it; 2 Move copies it (server-side, with metadata and tags) to <PostProcessBucket or the same bucket>/<PostProcessPrefix><key> and deletes the original; 3 Tag adds PostProcessTag to the object's tags (other tags are kept). Delete and Move only act on the version that was read: if the object was changed meanwhile they fail with HTTP 412 and the object is kept. A post-processing error raises an error; the data is already committed and the object is read again by the next run.
• When Output or the commit fails, the batch fails (sinks roll back) and the object is not post-processed. With ErrorBucket or ErrorPrefix set it is moved to <ErrorBucket or the same bucket>/<ErrorPrefix><key> (a failed move is logged). The error is raised in both cases.

Notes

• Downstream components must consume the stream before Output returns (decoders do); it is closed afterwards. The stream is forward-only (not seekable).
• S3 has no setting like Blob reader's AllowModifications: one GET always returns one consistent version.
• Moving within the listed bucket: give the target a prefix outside the enumerator's Prefix (for example list in/, move to done/), or the next run lists the moved objects again. Tag + the enumerator's SkipTag keeps the objects in place; S3 allows at most 10 tags per object.
• AWS S3 checks the version atomically (If-Match on the delete, x-amz-copy-source-if-match on the copy). Some S3-compatible stores (MinIO) ignore If-Match on deletes; there a conditional HEAD request right before the delete does the check, which leaves a short window.
• In a versioned bucket Delete and Move leave a delete marker; the earlier versions stay.
• Objects in Glacier Flexible Retrieval or Deep Archive must be restored before they can be read.
• Store secrets as expression settings (=Secrets.AwsSecretKey).

Example

{ "Authentication": 1, "Region": "eu-central-1", "PostProcessing": 2, "PostProcessPrefix": "done/", "ErrorPrefix": "error/" }

Timer → S3 enumerator (Prefix in/) → S3 reader → CSV decoder → Table writer (bucket and key from the enumerator)

Timer → S3 reader (BucketName, Key set) → CSV decoder → Table writer; (NoData) Log