forked from pulp/pulp_container
-
Notifications
You must be signed in to change notification settings - Fork 0
/
redirects.py
151 lines (127 loc) · 5.32 KB
/
redirects.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
from django.conf import settings
from django.core.exceptions import ObjectDoesNotExist
from django.http import Http404
from django.shortcuts import redirect
from pulp_container.app.utils import get_accepted_media_types
from pulp_container.constants import BLOB_CONTENT_TYPE, MEDIA_TYPE
class CommonRedirects:
"""
A class that serves for common redirects which target the content app.
"""
def __init__(self, distribution, path, request):
"""
Initialize fields which are required for performing specific redirects.
"""
self.distribution = distribution
self.path = path
self.request = request
def redirect_to_content_app(self, content_type, content_id):
"""
Redirect to the content app.
"""
return self.distribution.redirect_to_content_app(
f"{settings.CONTENT_ORIGIN}/pulp/container/{self.path}/{content_type}/{content_id}"
)
class FileStorageRedirects(CommonRedirects):
"""
A class which contains methods used for redirecting to the default django's file storage.
"""
def issue_tag_redirect(self, tag):
"""
Issue a redirect for the passed tag.
"""
return self.redirect_to_content_app("manifests", tag.name)
def issue_manifest_redirect(self, manifest):
"""
Issue a redirect for the passed manifest.
"""
return self.redirect_to_content_app("manifests", manifest.digest)
def issue_blob_redirect(self, blob):
"""
Issue a redirect for the passed blob.
"""
return self.redirect_to_content_app("blobs", blob.digest)
class S3StorageRedirects(CommonRedirects):
"""
A class that implements methods for the direct retrieval of manifest objects.
"""
def issue_tag_redirect(self, tag):
"""
Issue a redirect or perform a schema conversion if an accepted media type requires it.
"""
manifest_media_type = tag.tagged_manifest.media_type
if manifest_media_type in get_accepted_media_types(self.request.headers):
return self.redirect_to_artifact(tag.name, tag.tagged_manifest, manifest_media_type)
elif manifest_media_type == MEDIA_TYPE.MANIFEST_V1:
return self.redirect_to_artifact(
tag.name, tag.tagged_manifest, MEDIA_TYPE.MANIFEST_V1_SIGNED
)
else:
# execute the schema conversion
return self.redirect_to_content_app("manifests", tag.name)
def issue_manifest_redirect(self, manifest):
"""
Directly redirect to an associated manifest's artifact.
"""
return self.redirect_to_artifact(manifest.digest, manifest, manifest.media_type)
def redirect_to_artifact(self, content_name, manifest, manifest_media_type):
"""
Search for the passed manifest's artifact and issue a redirect.
"""
try:
artifact = manifest._artifacts.get()
except ObjectDoesNotExist:
raise Http404(f"An artifact for '{content_name}' was not found")
return self.redirect_to_object_storage(artifact, manifest_media_type)
def issue_blob_redirect(self, blob):
"""
Redirect to the passed blob or stream content when an associated artifact is not present.
"""
try:
artifact = blob._artifacts.get()
except ObjectDoesNotExist:
return self.redirect_to_content_app("blobs", blob.digest)
return self.redirect_to_object_storage(artifact, BLOB_CONTENT_TYPE)
def redirect_to_object_storage(self, artifact, return_media_type):
"""
Redirect to the passed artifact's file stored in the S3 storage.
"""
filename = f"sha256:{artifact.sha256}"
parameters = {
"ResponseContentType": return_media_type,
"ResponseContentDisposition": f"attachment;filename={filename}",
}
content_url = artifact.file.storage.url(
artifact.file.name, parameters=parameters, http_method=self.request.method
)
return redirect(content_url)
class AzureStorageRedirects(S3StorageRedirects):
"""
A class that implements methods for the direct retrieval of manifest objects.
"""
def redirect_to_object_storage(self, artifact, return_media_type):
"""
Redirect to the passed artifact's file stored in the Azure storage.
"""
filename = f"sha256:{artifact.sha256}"
parameters = {
"content_type": return_media_type,
"content_disposition": f"attachment;filename={filename}",
}
content_url = artifact.file.storage.url(artifact.file.name, parameters=parameters)
return redirect(content_url)
class GCloudStorageRedirects(S3StorageRedirects):
"""
A class that implements methods for the direct retrieval of manifest objects.
"""
def redirect_to_object_storage(self, artifact, return_media_type):
"""
Redirect to the passed artifact's file stored in the Azure storage.
"""
filename = f"sha256:{artifact.sha256}"
parameters = {
"content_type": return_media_type,
"response_disposition": f"attachment;filename={filename}",
}
content_url = artifact.file.storage.url(artifact.file.name, parameters=parameters)
return redirect(content_url)