Skip to content

S3 Writer (API)

Writes JSON data to the configured S3 bucket and logs success/failure.

Overview

  • Constructor requires: logger, s3_client (boto3), and bucket_name.
  • Validates bucket_name is set; otherwise raises ValueError.
  • Method write_data_to_s3(file_to_update, data) uploads JSON to s3://<bucket>/<file_to_update>.

Quick Start

import boto3, json
from s3writer import S3Writer
from logger import wrapped_logging

logger = wrapped_logging(False)
s3 = boto3.client("s3")

writer = S3Writer(logger, s3, bucket_name="<bucket>")

payload = {"alice": ["alice@org.com"], "bob": ["bob@org.com"]}
writer.write_data_to_s3("AddressBook/addressBookUsernameKey.json", payload)

Errors

  • Raises Exception if file_to_update or data is None.
  • Logs errors and re-raises on S3 failures (e.g., access denied).

Reference

This file contains the S3Writer class to weekly upload the GitHub usernames and ONS emails to S3

Typical usage example:

s3writer = S3Writer(logger)
s3writer.write_data_to_s3(file_to_update, data) # These are the filename of the updated file and its contents respectively

S3Writer

A class for uploading updated GitHub username and ONS emails to an AWS S3 Bucket

This class provides a setup and then can be used for all uploading throughout the programs lifecycle.

Atrributes

s3_resrouce: The established 'session' or connection to AWS servers bucket_name: The name of the bucket to store the new JSON files logger: The variable which connects to the logger class

Methods:

Name Description
write_data_to_s3

Allows the program to connect to the S3 bucket and upload the JSON
Source code in src/s3writer.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
class S3Writer:
    """
    A class for uploading updated GitHub username and ONS emails to an AWS S3 Bucket

    This class provides a setup and then can be used for all uploading throughout the programs lifecycle.

    Atrributes:
        s3_resrouce: The established 'session' or connection to AWS servers
        bucket_name: The name of the bucket to store the new JSON files
        logger: The variable which connects to the logger class

    Methods:
        write_data_to_s3: Allows the program to connect to the S3 bucket and upload the JSON
    """

    def __init__(self, logger, s3_client, bucket_name):
        """
        Initialises the S3Writer.
        """
        self.logger = logger
        self.s3_client = s3_client

        # Load bucket name from environment variable
        self.bucket_name = bucket_name
        if not self.bucket_name:
            raise ValueError(
                "S3_BUCKET_NAME environment variable is not set."
                "Please create a .env file with S3_BUCKET_NAME=your-bucket-name"
            )

    def write_data_to_s3(
        self, file_to_update: str | None, data: dict[str, Any] | str | None
    ):
        """
        Writes the data to a specific filename within the specificed s3 bucket

        Args:
            file_to_update: Name of the file to update within S3
            data: Contents of the new and updated file

            Raises:
            Exception: If filename or data is empty
            Exception: If S3 update fails
        """

        # Ensure that the arguments are not None
        if file_to_update is None or data is None:
            message = f"filename or data is empty. filename: {'empty' if file_to_update is None else 'filled'}, data: {'empty' if data is None else 'filled'}"
            self.logger.log_error(message)
            raise Exception(message)

        # Convert dict to JSON string if needed
        if isinstance(data, dict):
            data_str = json.dumps(data, indent=2)
        else:
            data_str = data

        # Upload the file to S3 within the bucket directly
        key = f"{file_to_update}"

        try:

            self.s3_client.put_object(
                Bucket=self.bucket_name,
                Key=key,
                Body=(
                    data_str.encode("utf-8")
                    if isinstance(data_str, str)
                    else json.dumps(data_str).encode("utf-8")
                ),
                ContentType="application/json",
            )

        except Exception as error:
            self.logger.log_error(
                f"Unable to upload updated username and email data to S3, {error}"
            )
            raise error
        else:
            self.logger.log_info(
                "Successfully uploaded updated username and email data to S3"
            )

__init__(logger, s3_client, bucket_name)

Initialises the S3Writer.

Source code in src/s3writer.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def __init__(self, logger, s3_client, bucket_name):
    """
    Initialises the S3Writer.
    """
    self.logger = logger
    self.s3_client = s3_client

    # Load bucket name from environment variable
    self.bucket_name = bucket_name
    if not self.bucket_name:
        raise ValueError(
            "S3_BUCKET_NAME environment variable is not set."
            "Please create a .env file with S3_BUCKET_NAME=your-bucket-name"
        )

write_data_to_s3(file_to_update, data)

Writes the data to a specific filename within the specificed s3 bucket

Parameters:

Name Type Description Default
file_to_update str | None

Name of the file to update within S3

 required
data dict[str, Any] | str | None

Contents of the new and updated file

 required
Raises
required
Exception

If filename or data is empty

 required
Exception

If S3 update fails

 required
Source code in src/s3writer.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def write_data_to_s3(
    self, file_to_update: str | None, data: dict[str, Any] | str | None
):
    """
    Writes the data to a specific filename within the specificed s3 bucket

    Args:
        file_to_update: Name of the file to update within S3
        data: Contents of the new and updated file

        Raises:
        Exception: If filename or data is empty
        Exception: If S3 update fails
    """

    # Ensure that the arguments are not None
    if file_to_update is None or data is None:
        message = f"filename or data is empty. filename: {'empty' if file_to_update is None else 'filled'}, data: {'empty' if data is None else 'filled'}"
        self.logger.log_error(message)
        raise Exception(message)

    # Convert dict to JSON string if needed
    if isinstance(data, dict):
        data_str = json.dumps(data, indent=2)
    else:
        data_str = data

    # Upload the file to S3 within the bucket directly
    key = f"{file_to_update}"

    try:

        self.s3_client.put_object(
            Bucket=self.bucket_name,
            Key=key,
            Body=(
                data_str.encode("utf-8")
                if isinstance(data_str, str)
                else json.dumps(data_str).encode("utf-8")
            ),
            ContentType="application/json",
        )

    except Exception as error:
        self.logger.log_error(
            f"Unable to upload updated username and email data to S3, {error}"
        )
        raise error
    else:
        self.logger.log_info(
            "Successfully uploaded updated username and email data to S3"
        )