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
|
######################################################################
#
# File: b2sdk/v1/session.py
#
# Copyright 2021 Backblaze Inc. All Rights Reserved.
#
# License https://www.backblaze.com/using_b2_code.html
#
######################################################################
from __future__ import annotations
from b2sdk import v2
from b2sdk.v2.exception import InvalidArgument
from .account_info import SqliteAccountInfo
# Override to use legacy signature of account_info.set_auth_data, especially the minimum_part_size argument
# and to accept old-style raw_api argument
class B2Session(v2.B2Session):
SQLITE_ACCOUNT_INFO_CLASS = staticmethod(SqliteAccountInfo)
def __init__(
self,
account_info=None,
cache=None,
raw_api: v2.B2RawHTTPApi = None,
api_config: v2.B2HttpApiConfig | None = None,
):
if raw_api is not None and api_config is not None:
raise InvalidArgument(
'raw_api,api_config', 'Provide at most one of: raw_api, api_config'
)
if api_config is None:
api_config = v2.DEFAULT_HTTP_API_CONFIG
super().__init__(account_info=account_info, cache=cache, api_config=api_config)
if raw_api is not None:
self.raw_api = raw_api
def authorize_account(self, realm, application_key_id, application_key):
"""
Perform account authorization.
:param str realm: a realm to authorize account in (usually just "production")
:param str application_key_id: :term:`application key ID`
:param str application_key: user's :term:`application key`
"""
# Authorize
realm_url = self.account_info.REALM_URLS.get(realm, realm)
response = self.raw_api.authorize_account(realm_url, application_key_id, application_key)
account_id = response['accountId']
storage_api_info = response['apiInfo']['storageApi']
# `allowed` object has been deprecated in the v3 of the API, but we still
# construct it artificially to avoid changes in all the reliant parts.
allowed = {
'bucketId': storage_api_info['bucketId'],
'bucketName': storage_api_info['bucketName'],
'capabilities': storage_api_info['capabilities'],
'namePrefix': storage_api_info['namePrefix'],
}
# Clear the cache if new account has been used
if not self.account_info.is_same_account(account_id, realm):
self.cache.clear()
# Store the auth data
self.account_info.set_auth_data(
account_id=account_id,
auth_token=response['authorizationToken'],
api_url=storage_api_info['apiUrl'],
download_url=storage_api_info['downloadUrl'],
minimum_part_size=storage_api_info['recommendedPartSize'],
application_key=application_key,
realm=realm,
s3_api_url=storage_api_info['s3ApiUrl'],
allowed=allowed,
application_key_id=application_key_id,
)
|
