Skip to content

Views

S3GetParamsView

Bases: GenericViewSet

View for getting params for s3 to upload file to S3.

Source code in saritasa_s3_tools/django/views.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
 81
 82
 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
class S3GetParamsView(viewsets.GenericViewSet):
    """View for getting params for s3 to upload file to S3."""

    serializer_class = serializers.S3RequestParamsSerializer
    permission_classes = (permissions.AllowAny,)
    filter_backends = ()
    pagination_class = None

    @decorators.action(
        methods=["POST"],
        url_path="get-params",
        url_name="get-params",
        detail=False,
    )
    def get_params(
        self,
        request: Request,
    ) -> response.Response:
        """Get parameters for upload to S3 bucket.

        Current endpoint returns all required for s3 upload data,
        which should be later sent to `url` as `form-data` url with
        'file'. Workflow: First, you make request to this endpoint. Then send
        response data to `url` via `POST` as form-data with file included. In
        response you will get an url which you can use in API for value of file
        related fields like avatar for example.

        """
        serializer = self.serializer_class(
            context_request=request,
            data=request.data,
        )
        serializer.is_valid(raise_exception=True)
        s3_client = self.get_s3_client()
        params = s3_client.generate_params(
            filename=serializer.data["filename"],  # type: ignore
            config=serializer.data["config"],  # type: ignore
            content_type=serializer.data["content_type"],  # type: ignore
            extra_metadata=self.get_extra_meta_data(user=request.user),
        )
        return response.Response(
            status=status.HTTP_200_OK,
            data=serializers.S3UploadSerializer(
                instance=dataclasses.asdict(params),
            ).data,
        )

    @decorators.action(
        methods=["GET"],
        url_path="list-configs",
        url_name="list-configs",
        detail=False,
    )
    def list_configs(
        self,
        request: Request,
    ) -> response.Response:
        """List all configs for s3 upload."""
        return response.Response(
            status=status.HTTP_200_OK,
            data=serializers.S3ConfigSerializer(
                instance=map(
                    dataclasses.asdict,
                    configs.S3FileTypeConfig.configs.values(),
                ),
                many=True,
            ).data,
        )

    @decorators.action(
        methods=["GET"],
        url_path="retrieve-config/(?P<name>[^/.]+)",
        url_name="retrieve-config",
        detail=False,
    )
    def retrieve_config(
        self,
        request: Request,
        name: str,
    ) -> response.Response:
        """Retrieve config for s3 upload."""
        if name not in configs.S3FileTypeConfig.configs:
            raise exceptions.NotFound
        return response.Response(
            status=status.HTTP_200_OK,
            data=serializers.S3ConfigSerializer(
                instance=dataclasses.asdict(
                    configs.S3FileTypeConfig.configs[name],
                ),
            ).data,
        )

    def get_s3_client(self) -> client.S3Client:
        """Get s3 client for params generation."""
        return shortcuts.get_s3_client()

    def get_extra_meta_data(
        self,
        user: typing.Any,
    ) -> dict[str, str]:
        """Extend meta data for file."""
        return {
            "user-id": str(user.pk),
        }

get_extra_meta_data(user)

Extend meta data for file.

Source code in saritasa_s3_tools/django/views.py 
115
116
117
118
119
120
121
122
def get_extra_meta_data(
    self,
    user: typing.Any,
) -> dict[str, str]:
    """Extend meta data for file."""
    return {
        "user-id": str(user.pk),
    }

get_params(request)

Get parameters for upload to S3 bucket.

Current endpoint returns all required for s3 upload data, which should be later sent to url as form-data url with 'file'. Workflow: First, you make request to this endpoint. Then send response data to url via POST as form-data with file included. In response you will get an url which you can use in API for value of file related fields like avatar for example.

Source code in saritasa_s3_tools/django/views.py 
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
@decorators.action(
    methods=["POST"],
    url_path="get-params",
    url_name="get-params",
    detail=False,
)
def get_params(
    self,
    request: Request,
) -> response.Response:
    """Get parameters for upload to S3 bucket.

    Current endpoint returns all required for s3 upload data,
    which should be later sent to `url` as `form-data` url with
    'file'. Workflow: First, you make request to this endpoint. Then send
    response data to `url` via `POST` as form-data with file included. In
    response you will get an url which you can use in API for value of file
    related fields like avatar for example.

    """
    serializer = self.serializer_class(
        context_request=request,
        data=request.data,
    )
    serializer.is_valid(raise_exception=True)
    s3_client = self.get_s3_client()
    params = s3_client.generate_params(
        filename=serializer.data["filename"],  # type: ignore
        config=serializer.data["config"],  # type: ignore
        content_type=serializer.data["content_type"],  # type: ignore
        extra_metadata=self.get_extra_meta_data(user=request.user),
    )
    return response.Response(
        status=status.HTTP_200_OK,
        data=serializers.S3UploadSerializer(
            instance=dataclasses.asdict(params),
        ).data,
    )

get_s3_client()

Get s3 client for params generation.

Source code in saritasa_s3_tools/django/views.py 
111
112
113
def get_s3_client(self) -> client.S3Client:
    """Get s3 client for params generation."""
    return shortcuts.get_s3_client()

list_configs(request)

List all configs for s3 upload.

Source code in saritasa_s3_tools/django/views.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@decorators.action(
    methods=["GET"],
    url_path="list-configs",
    url_name="list-configs",
    detail=False,
)
def list_configs(
    self,
    request: Request,
) -> response.Response:
    """List all configs for s3 upload."""
    return response.Response(
        status=status.HTTP_200_OK,
        data=serializers.S3ConfigSerializer(
            instance=map(
                dataclasses.asdict,
                configs.S3FileTypeConfig.configs.values(),
            ),
            many=True,
        ).data,
    )

retrieve_config(request, name)

Retrieve config for s3 upload.

Source code in saritasa_s3_tools/django/views.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
@decorators.action(
    methods=["GET"],
    url_path="retrieve-config/(?P<name>[^/.]+)",
    url_name="retrieve-config",
    detail=False,
)
def retrieve_config(
    self,
    request: Request,
    name: str,
) -> response.Response:
    """Retrieve config for s3 upload."""
    if name not in configs.S3FileTypeConfig.configs:
        raise exceptions.NotFound
    return response.Response(
        status=status.HTTP_200_OK,
        data=serializers.S3ConfigSerializer(
            instance=dataclasses.asdict(
                configs.S3FileTypeConfig.configs[name],
            ),
        ).data,
    )