--- id: aws--s3-reader title: "S3 reader" summary: "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`." audience: user keywords: "input, output, nodata, authentication, accesskeyid, secretaccesskey, sessiontoken, rolearn, externalid, stsserviceurl, region, serviceurl, forcepathstyle, bucketname, key, postprocessing, postprocessbucket, postprocessprefix, postprocesstag, errorbucket, errorprefix" --- # 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 `//`; 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:///` (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` = `/`, `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 `/` 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 `/` (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 ```json { "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