b2sdk._internal.raw_api – B2 raw api wrapper

class b2sdk._internal.raw_api.MetadataDirectiveMode(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Mode of handling metadata when copying a file

COPY = 401

copy metadata from the source file

REPLACE = 402

ignore the source file metadata and set it to provided values

class b2sdk._internal.raw_api.LifecycleRule[source]

Bases: TypedDict

Lifecycle Rule.

External documentation: B2 Cloud Storage Lifecycle Rules.

fileNamePrefix: str
daysFromHidingToDeleting: Optional[int]
daysFromUploadingToHiding: Optional[int]
class b2sdk._internal.raw_api.NameValueDict[source]

Bases: TypedDict

name: str
value: str
class b2sdk._internal.raw_api.NotificationTargetConfiguration[source]

Bases: TypedDict

Notification Target Configuration.

hmacSha256SigningSecret, if present, has to be a string of 32 alphanumeric characters.

targetType: Literal['webhook']
url: str
customHeaders: Optional[list[NameValueDict]]
hmacSha256SigningSecret: Optional[str]
class b2sdk._internal.raw_api.NotificationRule[source]

Bases: _NotificationRule

Notification Rule.

When creating or modifying a notification rule, isSuspended and suspensionReason are ignored.

isSuspended: bool
eventTypes: list[Literal['b2:ObjectCreated:*', 'b2:ObjectCreated:Upload', 'b2:ObjectCreated:MultipartUpload', 'b2:ObjectCreated:Copy', 'b2:ObjectCreated:Replica', 'b2:ObjectCreated:MultipartReplica', 'b2:ObjectDeleted:*', 'b2:ObjectDeleted:Delete', 'b2:ObjectDeleted:LifecycleRule', 'b2:HideMarkerCreated:*', 'b2:HideMarkerCreated:Hide', 'b2:HideMarkerCreated:LifecycleRule']]
isEnabled: bool
name: str
objectNamePrefix: str
targetConfiguration: NotificationTargetConfiguration
suspensionReason: str
class b2sdk._internal.raw_api.NotificationRuleResponse[source]

Bases: _NotificationRule

isSuspended: bool
eventTypes: list[Literal['b2:ObjectCreated:*', 'b2:ObjectCreated:Upload', 'b2:ObjectCreated:MultipartUpload', 'b2:ObjectCreated:Copy', 'b2:ObjectCreated:Replica', 'b2:ObjectCreated:MultipartReplica', 'b2:ObjectDeleted:*', 'b2:ObjectDeleted:Delete', 'b2:ObjectDeleted:LifecycleRule', 'b2:HideMarkerCreated:*', 'b2:HideMarkerCreated:Hide', 'b2:HideMarkerCreated:LifecycleRule']]
isEnabled: bool
name: str
objectNamePrefix: str
targetConfiguration: NotificationTargetConfiguration
suspensionReason: str
b2sdk._internal.raw_api.notification_rule_response_to_request(rule)[source]

Convert NotificationRuleResponse to NotificationRule.

Parameters:

rule (NotificationRuleResponse) –

Return type:

NotificationRule

class b2sdk._internal.raw_api.AbstractRawApi[source]

Bases: object

Direct access to the B2 web apis.

abstract authorize_account(realm_url, application_key_id, application_key)[source]
abstract cancel_large_file(api_url, account_auth_token, file_id)[source]
abstract copy_file(api_url, account_auth_token, source_file_id, new_file_name, bytes_range=None, metadata_directive=None, content_type=None, file_info=None, destination_bucket_id=None, destination_server_side_encryption=None, source_server_side_encryption=None, file_retention=None, legal_hold=None)[source]
Parameters:
abstract copy_part(api_url, account_auth_token, source_file_id, large_file_id, part_number, bytes_range=None, destination_server_side_encryption=None, source_server_side_encryption=None)[source]
Parameters:
abstract create_bucket(api_url, account_auth_token, account_id, bucket_name, bucket_type, bucket_info=None, cors_rules=None, lifecycle_rules=None, default_server_side_encryption=None, is_file_lock_enabled=None, replication=None)[source]
Parameters:
abstract create_key(api_url, account_auth_token, account_id, capabilities, key_name, valid_duration_seconds, bucket_id, name_prefix)[source]
abstract download_file_from_url(account_auth_token_or_none, url, range_=None, encryption=None)[source]
Parameters:

encryption (Optional[EncryptionSetting]) –

abstract delete_key(api_url, account_auth_token, application_key_id)[source]
abstract delete_bucket(api_url, account_auth_token, account_id, bucket_id)[source]
abstract delete_file_version(api_url, account_auth_token, file_id, file_name, bypass_governance=False)[source]
Parameters:

bypass_governance (bool) –

abstract finish_large_file(api_url, account_auth_token, file_id, part_sha1_array)[source]
abstract get_download_authorization(api_url, account_auth_token, bucket_id, file_name_prefix, valid_duration_in_seconds)[source]
abstract get_file_info_by_id(api_url, account_auth_token, file_id)[source]
Parameters:

  • api_url (str) –

  • account_auth_token (str) –

  • file_id (str) –

Return type:

dict[str, Any]

abstract get_file_info_by_name(download_url, account_auth_token, bucket_name, file_name)[source]
Parameters:

  • download_url (str) –

  • account_auth_token (str) –

  • bucket_name (str) –

  • file_name (str) –

Return type:

dict[str, Any]

abstract get_upload_url(api_url, account_auth_token, bucket_id)[source]
abstract get_upload_part_url(api_url, account_auth_token, file_id)[source]
abstract hide_file(api_url, account_auth_token, bucket_id, file_name)[source]
abstract list_buckets(api_url, account_auth_token, account_id, bucket_id=None, bucket_name=None)[source]
abstract list_file_names(api_url, account_auth_token, bucket_id, start_file_name=None, max_file_count=None, prefix=None)[source]
abstract list_file_versions(api_url, account_auth_token, bucket_id, start_file_name=None, start_file_id=None, max_file_count=None, prefix=None)[source]
abstract list_keys(api_url, account_auth_token, account_id, max_key_count=None, start_application_key_id=None)[source]
abstract list_parts(api_url, account_auth_token, file_id, start_part_number, max_part_count)[source]
abstract list_unfinished_large_files(api_url, account_auth_token, bucket_id, start_file_id=None, max_file_count=None, prefix=None)[source]
abstract start_large_file(api_url, account_auth_token, bucket_id, file_name, content_type, file_info, server_side_encryption=None, file_retention=None, legal_hold=None)[source]
Parameters:
abstract update_bucket(api_url, account_auth_token, account_id, bucket_id, bucket_type=None, bucket_info=None, cors_rules=None, lifecycle_rules=None, if_revision_is=None, default_server_side_encryption=None, default_retention=None, replication=None, is_file_lock_enabled=None)[source]
Parameters:
abstract update_file_retention(api_url, account_auth_token, file_id, file_name, file_retention, bypass_governance=False)[source]
Parameters:
classmethod 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=None)[source]
Parameters:
Return type:

dict

abstract upload_file(upload_url, upload_auth_token, file_name, content_length, content_type, content_sha1, file_info, data_stream, server_side_encryption=None, file_retention=None, legal_hold=None, custom_upload_timestamp=None)[source]
Parameters:
abstract upload_part(upload_url, upload_auth_token, part_number, content_length, sha1_sum, input_stream, server_side_encryption=None)[source]
Parameters:

server_side_encryption (Optional[EncryptionSetting]) –

get_download_url_by_id(download_url, file_id)[source]
get_download_url_by_name(download_url, bucket_name, file_name)[source]
abstract set_bucket_notification_rules(api_url, account_auth_token, bucket_id, rules)[source]
Parameters:
Return type:

list[NotificationRuleResponse]

abstract get_bucket_notification_rules(api_url, account_auth_token, bucket_id)[source]
Parameters:

  • api_url (str) –

  • account_auth_token (str) –

  • bucket_id (str) –

Return type:

list[NotificationRuleResponse]

class b2sdk._internal.raw_api.B2RawHTTPApi(b2_http)[source]

Bases: AbstractRawApi

Provide access to the B2 web APIs, exactly as they are provided by b2.

Requires that you provide all necessary URLs and auth tokens for each call.

Each API call decodes the returned JSON and returns a dict.

For details on what each method does, see the B2 docs:

https://www.backblaze.com/b2/docs/

This class is intended to be a super-simple, very thin layer on top of the HTTP calls. It can be mocked-out for testing higher layers. And this class can be tested by exercising each call just once, which is relatively quick.

__init__(b2_http)[source]
authorize_account(realm_url, application_key_id, application_key)[source]
cancel_large_file(api_url, account_auth_token, file_id)[source]
create_bucket(api_url, account_auth_token, account_id, bucket_name, bucket_type, bucket_info=None, cors_rules=None, lifecycle_rules=None, default_server_side_encryption=None, is_file_lock_enabled=None, replication=None)[source]
Parameters:
create_key(api_url, account_auth_token, account_id, capabilities, key_name, valid_duration_seconds, bucket_id, name_prefix)[source]
delete_bucket(api_url, account_auth_token, account_id, bucket_id)[source]
delete_file_version(api_url, account_auth_token, file_id, file_name, bypass_governance=False)[source]
Parameters:

bypass_governance (bool) –

delete_key(api_url, account_auth_token, application_key_id)[source]
download_file_from_url(account_auth_token_or_none, url, range_=None, encryption=None)[source]

Issue a streaming request for download of a file, potentially authorized.

Parameters:

  • account_auth_token_or_none (str) – an optional account auth token to pass in

  • url (str) – the full URL to download from

  • range (tuple) – two-element tuple for http Range header

  • encryption (Optional[EncryptionSetting]) – encryption settings for downloading

  • encryption –

Returns:

b2_http response

finish_large_file(api_url, account_auth_token, file_id, part_sha1_array)[source]
get_download_authorization(api_url, account_auth_token, bucket_id, file_name_prefix, valid_duration_in_seconds)[source]
get_file_info_by_id(api_url, account_auth_token, file_id)[source]
Parameters:

  • api_url (str) –

  • account_auth_token (str) –

  • file_id (str) –

Return type:

dict[str, Any]

get_file_info_by_name(download_url, account_auth_token, bucket_name, file_name)[source]
Parameters:

  • download_url (str) –

  • account_auth_token (str) –

  • bucket_name (str) –

  • file_name (str) –

Return type:

dict[str, Any]

get_upload_url(api_url, account_auth_token, bucket_id)[source]
get_upload_part_url(api_url, account_auth_token, file_id)[source]
hide_file(api_url, account_auth_token, bucket_id, file_name)[source]
list_buckets(api_url, account_auth_token, account_id, bucket_id=None, bucket_name=None)[source]
list_file_names(api_url, account_auth_token, bucket_id, start_file_name=None, max_file_count=None, prefix=None)[source]
list_file_versions(api_url, account_auth_token, bucket_id, start_file_name=None, start_file_id=None, max_file_count=None, prefix=None)[source]
list_keys(api_url, account_auth_token, account_id, max_key_count=None, start_application_key_id=None)[source]
list_parts(api_url, account_auth_token, file_id, start_part_number, max_part_count)[source]
list_unfinished_large_files(api_url, account_auth_token, bucket_id, start_file_id=None, max_file_count=None, prefix=None)[source]
start_large_file(api_url, account_auth_token, bucket_id, file_name, content_type, file_info, server_side_encryption=None, file_retention=None, legal_hold=None, custom_upload_timestamp=None)[source]
Parameters:
update_bucket(api_url, account_auth_token, account_id, bucket_id, bucket_type=None, bucket_info=None, cors_rules=None, lifecycle_rules=None, if_revision_is=None, default_server_side_encryption=None, default_retention=None, replication=None, is_file_lock_enabled=None)[source]
Parameters:
update_file_retention(api_url, account_auth_token, file_id, file_name, file_retention, bypass_governance=False)[source]
Parameters:
Parameters:

legal_hold (LegalHold) –

check_b2_filename(filename)[source]

Raise an appropriate exception with details if the filename is unusable.

See https://www.backblaze.com/b2/docs/files.html for the rules.

Parameters:

filename – a proposed filename in unicode

Returns:

None if the filename is usable

upload_file(upload_url, upload_auth_token, file_name, content_length, content_type, content_sha1, file_info, data_stream, server_side_encryption=None, file_retention=None, legal_hold=None, custom_upload_timestamp=None)[source]

Upload one, small file to b2.

Parameters:

  • upload_url – the upload_url from b2_authorize_account

  • upload_auth_token – the auth token from b2_authorize_account

  • file_name – the name of the B2 file

  • content_length – number of bytes in the file

  • content_type – MIME type

  • content_sha1 – hex SHA1 of the contents of the file

  • file_info (dict) – extra file info to upload

  • data_stream – a file like object from which the contents of the file can be read

  • server_side_encryption (Optional[EncryptionSetting]) – encryption setting for the file

  • file_retention (Optional[FileRetentionSetting]) – retention setting for the file

  • legal_hold (Optional[LegalHold]) – legal hold setting for the file

  • custom_upload_timestamp (Optional[int]) – custom upload timestamp for the file

Returns:

upload_part(upload_url, upload_auth_token, part_number, content_length, content_sha1, data_stream, server_side_encryption=None)[source]
Parameters:

server_side_encryption (Optional[EncryptionSetting]) –

copy_file(api_url, account_auth_token, source_file_id, new_file_name, bytes_range=None, metadata_directive=None, content_type=None, file_info=None, destination_bucket_id=None, destination_server_side_encryption=None, source_server_side_encryption=None, file_retention=None, legal_hold=None)[source]
Parameters:
copy_part(api_url, account_auth_token, source_file_id, large_file_id, part_number, bytes_range=None, destination_server_side_encryption=None, source_server_side_encryption=None)[source]
Parameters:
set_bucket_notification_rules(api_url, account_auth_token, bucket_id, rules)[source]
Parameters:
Return type:

list[NotificationRuleResponse]

get_bucket_notification_rules(api_url, account_auth_token, bucket_id)[source]
Parameters:

  • api_url (str) –

  • account_auth_token (str) –

  • bucket_id (str) –

Return type:

list[NotificationRuleResponse]