Skip to content

Factory

S3FileField

Bases: LazyAttribute

Generate file and upload to s3.

Source code in saritasa_s3_tools/factory.py 
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
class S3FileField(factory.LazyAttribute):
    """Generate file and upload to s3."""

    _boto3_client: mypy_boto3_s3.S3Client | None = None

    def __init__(
        self,
        s3_config: str,
        s3_region: client.RegionGetter | str,
        bucket: BucketGetter | str,
        access_key_getter: client.AccessKeyGetter,
        s3_endpoint_url_getter: client.S3EndpointUrlGetter | None = None,
        filename: str = "example.txt",
        data: bytes = b"Test",
    ) -> None:
        self.access_key_getter = access_key_getter
        self.s3_endpoint_url_getter = s3_endpoint_url_getter
        self.bucket = bucket
        self.s3_region = s3_region
        self.filename = filename
        self.s3_config = s3_config
        self.data = data

        super().__init__(
            function=self._generate,
        )

    def _get_boto3(self) -> mypy_boto3_s3.S3Client:
        if not self._boto3_client:
            self._boto3_client = client.get_boto3_s3_client(
                region=self.s3_region,
                s3_endpoint_url_getter=self.s3_endpoint_url_getter,
                access_key_getter=self.access_key_getter,
            )
        return self._boto3_client

    def _get_s3_client(self) -> client.S3Client:
        """Set up s3 client."""
        if isinstance(self.bucket, str):
            bucket = self.bucket
        if callable(self.bucket):
            bucket = self.bucket()
        return client.S3Client(
            boto3_client=self._get_boto3(),
            default_bucket=bucket,
        )

    def _generate_file_data(self) -> bytes:
        """Generate data for file."""
        return self.data

    def _generate(
        self,
        *args,  # noqa: ANN002
        **kwargs,
    ) -> str:
        """Generate file and upload it to s3."""
        return self._get_s3_client().upload_file(
            filename=self.filename,
            file_obj=io.BytesIO(self._generate_file_data()),
            config=configs.S3FileTypeConfig.configs[self.s3_config],
        )

S3ImageFileField

Bases: S3FileField

Generate image file and upload to s3.

Source code in saritasa_s3_tools/factory.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
class S3ImageFileField(S3FileField):
    """Generate image file and upload to s3."""

    def __init__(
        self,
        s3_config: str,
        s3_region: str,
        bucket: str,
        access_key_getter: client.AccessKeyGetter,
        s3_endpoint_url_getter: client.S3EndpointUrlGetter | None = None,
        filename: str = "example.jpg",
        data: bytes = b"",
        color: str = "",
        width: int = 0,
        height: int = 0,
    ) -> None:
        self.color = color
        self.width = width
        self.height = height
        super().__init__(
            s3_config=s3_config,
            s3_region=s3_region,
            bucket=bucket,
            access_key_getter=access_key_getter,
            s3_endpoint_url_getter=s3_endpoint_url_getter,
            data=data,
            filename=filename,
        )

    def _generate_file_data(self) -> bytes:
        """Generate image data for file."""
        width = self.width or self.height or 100
        height = self.height or self.width or 100
        color = self.color or random.choice(  # noqa: S311
            list(PIL.ImageColor.colormap.keys()),
        )

        thumb_io = io.BytesIO()
        with PIL.Image.new("RGB", (width, height), color) as thumb:
            thumb.save(thumb_io, format="JPEG")
        return thumb_io.getvalue()