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
|
######################################################################
#
# File: b2sdk/v2/raw_simulator.py
#
# Copyright 2023 Backblaze Inc. All Rights Reserved.
#
# License https://www.backblaze.com/using_b2_code.html
#
######################################################################
from __future__ import annotations
from b2sdk import _v3 as v3
from b2sdk.v2._compat import _file_infos_rename
class BucketSimulator(v3.BucketSimulator):
@_file_infos_rename
def upload_file(
self,
upload_id: str,
upload_auth_token: str,
file_name: str,
content_length: int,
content_type: str,
content_sha1: str,
file_info: dict,
data_stream,
server_side_encryption: v3.EncryptionSetting | None = None,
file_retention: v3.FileRetentionSetting | None = None,
legal_hold: v3.LegalHold | None = None,
custom_upload_timestamp: int | None = None,
cache_control: str | None = None,
*args,
**kwargs,
):
if cache_control is not None:
file_info['b2-cache-control'] = cache_control
return super().upload_file(
upload_id,
upload_auth_token,
file_name,
content_length,
content_type,
content_sha1,
file_info,
data_stream,
server_side_encryption,
file_retention,
legal_hold,
custom_upload_timestamp,
*args,
**kwargs,
)
class RawSimulator(v3.RawSimulator):
@classmethod
@_file_infos_rename
def get_upload_file_headers(
cls,
upload_auth_token: str,
file_name: str,
content_length: int,
content_type: str,
content_sha1: str,
file_info: dict,
server_side_encryption: v3.EncryptionSetting | None,
file_retention: v3.FileRetentionSetting | None,
legal_hold: v3.LegalHold | None,
custom_upload_timestamp: int | None = None,
cache_control: str | None = None,
*args,
**kwargs,
) -> dict:
if cache_control is not None:
file_info['b2-cache-control'] = cache_control
return super().get_upload_file_headers(
upload_auth_token,
file_name,
content_length,
content_type,
content_sha1,
file_info,
server_side_encryption,
file_retention,
legal_hold,
custom_upload_timestamp,
*args,
**kwargs,
)
@_file_infos_rename
def upload_file(
self,
upload_url: str,
upload_auth_token: str,
file_name: str,
content_length: int,
content_type: str,
content_sha1: str,
file_info: dict,
data_stream,
server_side_encryption: v3.EncryptionSetting | None = None,
file_retention: v3.FileRetentionSetting | None = None,
legal_hold: v3.LegalHold | None = None,
custom_upload_timestamp: int | None = None,
cache_control: str | None = None,
*args,
**kwargs,
):
if cache_control is not None:
file_info['b2-cache-control'] = cache_control
return super().upload_file(
upload_url,
upload_auth_token,
file_name,
content_length,
content_type,
content_sha1,
file_info,
data_stream,
server_side_encryption,
file_retention,
legal_hold,
custom_upload_timestamp,
*args,
**kwargs,
)
|
