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
|
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import uuid
from odoo import models, fields, _
from odoo.exceptions import UserError
from odoo.http import Stream
class CloudStorageAttachment(models.Model):
_inherit = 'ir.attachment'
_cloud_storage_upload_url_time_to_expiry = 300 # 300 seconds
_cloud_storage_download_url_time_to_expiry = 300 # 300 seconds
type = fields.Selection(
selection_add=[('cloud_storage', 'Cloud Storage')],
ondelete={'cloud_storage': 'set url'}
)
def _to_http_stream(self):
if (self.type == 'cloud_storage' and
self.env['res.config.settings']._get_cloud_storage_configuration()):
self.ensure_one()
info = self._generate_cloud_storage_download_info()
stream = Stream(type='url', url=info['url'])
if 'time_to_expiry' in info:
# cache the redirection until 10 seconds before the expiry
stream.max_age = max(info['time_to_expiry'] - 10, 0)
return stream
return super()._to_http_stream()
def _post_add_create(self, **kwargs):
super()._post_add_create(**kwargs)
if kwargs.get('cloud_storage'):
if not self.env['ir.config_parameter'].sudo().get_param('cloud_storage_provider'):
raise UserError(_('Cloud Storage is not enabled'))
for record in self:
record.write({
'raw': False,
'type': 'cloud_storage',
'url': record._generate_cloud_storage_url(),
})
def _generate_cloud_storage_blob_name(self):
"""
Generate a unique blob name for the attachment
:param attachment: an ir.attachment record
:return: A unique blob name str
"""
return f'{self.id}/{uuid.uuid4()}/{self.name}'
# Implement the following methods for each cloud storage provider.
def _generate_cloud_storage_url(self):
"""
Generate a cloud blob url without signature or token for the attachment.
This url is only used to identify the cloud blob.
:param attachment: an ir.attachment record
:return: A cloud blob url str
"""
raise NotImplementedError()
def _generate_cloud_storage_download_info(self):
"""
Generate the download info for the public client to directly download
the attachment's blob from the cloud storage.
:param attachment: an ir.attachment record
:return: An download_info dictionary containing:
* download_url: cloud storage url with permission to download the file
* time_to_expiry: the time in seconds before the download url expires
"""
raise NotImplementedError()
def _generate_cloud_storage_upload_info(self):
"""
Generate the upload info for the public client to directly upload a
file to the cloud storage.
:param attachment: an ir.attachment record
:return: An upload_info dictionary containing:
* upload_url: cloud storage url with permission to upload the file
* method: the request method used to upload the file
* response_status: the status of the response for a successful
upload request
* [Optionally] headers: a dictionary of headers to be added to the
upload request
"""
raise NotImplementedError()
|
