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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
######################################################################
#
# File: b2sdk/v2/bucket.py
#
# Copyright 2021 Backblaze Inc. All Rights Reserved.
#
# License https://www.backblaze.com/using_b2_code.html
#
######################################################################
from __future__ import annotations
import typing
from b2sdk import _v3 as v3
from b2sdk._v3.exception import BucketIdNotFound as v3BucketIdNotFound
from b2sdk.v2._compat import _file_infos_rename
from b2sdk._internal.http_constants import LIST_FILE_NAMES_MAX_LIMIT
from .exception import BucketIdNotFound
from .file_version import FileVersionFactory
if typing.TYPE_CHECKING:
from b2sdk._internal.utils import Sha1HexDigest
from b2sdk._internal.filter import Filter
from .file_version import FileVersion
# Overridden to raise old style BucketIdNotFound exception
class Bucket(v3.Bucket):
FILE_VERSION_FACTORY_CLASS = staticmethod(FileVersionFactory)
def get_fresh_state(self) -> Bucket:
try:
return super().get_fresh_state()
except v3BucketIdNotFound as e:
raise BucketIdNotFound(e.bucket_id)
@_file_infos_rename
def upload_bytes(
self,
data_bytes,
file_name,
content_type=None,
file_info: dict | None = None,
progress_listener=None,
encryption: v3.EncryptionSetting | None = None,
file_retention: v3.FileRetentionSetting | None = None,
legal_hold: v3.LegalHold | None = None,
large_file_sha1: Sha1HexDigest | None = None,
custom_upload_timestamp: int | None = None,
cache_control: str | None = None,
*args,
**kwargs,
):
return super().upload_bytes(
data_bytes,
file_name,
content_type,
file_info,
progress_listener,
encryption,
file_retention,
legal_hold,
large_file_sha1,
custom_upload_timestamp,
cache_control,
*args,
**kwargs,
)
@_file_infos_rename
def upload_local_file(
self,
local_file,
file_name,
content_type=None,
file_info: dict | None = None,
sha1_sum=None,
min_part_size=None,
progress_listener=None,
encryption: v3.EncryptionSetting | None = None,
file_retention: v3.FileRetentionSetting | None = None,
legal_hold: v3.LegalHold | None = None,
upload_mode: v3.UploadMode = v3.UploadMode.FULL,
custom_upload_timestamp: int | None = None,
cache_control: str | None = None,
*args,
**kwargs,
):
return super().upload_local_file(
local_file,
file_name,
content_type,
file_info,
sha1_sum,
min_part_size,
progress_listener,
encryption,
file_retention,
legal_hold,
upload_mode,
custom_upload_timestamp,
cache_control,
*args,
**kwargs,
)
def ls(
self,
folder_to_list: str = '',
latest_only: bool = True,
recursive: bool = False,
fetch_count: int | None = LIST_FILE_NAMES_MAX_LIMIT,
with_wildcard: bool = False,
filters: typing.Sequence[Filter] = (),
folder_to_list_can_be_a_file: bool = False,
**kwargs,
) -> typing.Iterable[tuple[FileVersion, str]]:
"""
Pretend that folders exist and yields the information about the files in a folder.
B2 has a flat namespace for the files in a bucket, but there is a convention
of using "/" as if there were folders. This method searches through the
flat namespace to find the files and "folders" that live within a given
folder.
When the `recursive` flag is set, lists all of the files in the given
folder, and all of its sub-folders.
:param folder_to_list: the name of the folder to list; must not start with "/".
Empty string means top-level folder
:param latest_only: when ``False`` returns info about all versions of a file,
when ``True``, just returns info about the most recent versions
:param recursive: if ``True``, list folders recursively
:param fetch_count: how many entries to list per API call or ``None`` to use the default. Acceptable values: 1 - 10000
:param with_wildcard: Accepts "*", "?", "[]" and "[!]" in folder_to_list, similarly to what shell does.
As of 1.19.0 it can only be enabled when recursive is also enabled.
Also, in this mode, folder_to_list is considered to be a filename or a pattern.
:param filters: list of filters to apply to the files returned by the server.
:param folder_to_list_can_be_a_file: if ``True``, folder_to_list can be a file, not just a folder
This enabled default behavior of b2sdk.v3.Bucket.ls, in which for all
paths that do not end with '/', first we try to check if file with this
exact name exists, and only if it does not then we try to list files with
this prefix.
:rtype: generator[tuple[b2sdk.v2.FileVersion, str]]
:returns: generator of (file_version, folder_name) tuples
.. note::
In case of `recursive=True`, folder_name is not returned.
"""
if (
not folder_to_list_can_be_a_file
and folder_to_list
and not folder_to_list.endswith('/')
and not with_wildcard
):
folder_to_list += '/'
yield from super().ls(
path=folder_to_list,
latest_only=latest_only,
recursive=recursive,
fetch_count=fetch_count,
with_wildcard=with_wildcard,
filters=filters,
**kwargs,
)
# Overridden to use old style Bucket
class BucketFactory(v3.BucketFactory):
BUCKET_CLASS = staticmethod(Bucket)
|
