Bunq update 7

bunq/sdk/client.py

@@ -284,9 +284,7 @@ def _fetch_response_id(self, response):
         if self.HEADER_RESPONSE_ID_LOWER_CASED in headers:
             return headers[self.HEADER_RESPONSE_ID_LOWER_CASED]
 
-        return exception.BunqException(
-            self._ERROR_COULD_NOT_DETERMINE_RESPONSE_ID_HEADER
-        )
+        return self._ERROR_COULD_NOT_DETERMINE_RESPONSE_ID_HEADER;
 
     def put(self, uri_relative, request_bytes, custom_headers):
         """

bunq/sdk/context.py

@@ -3,9 +3,11 @@
 import aenum
 from Cryptodome.PublicKey import RSA
 
-from bunq.sdk.model import core
 from bunq.sdk import security
+from bunq.sdk.exception import BunqException
 from bunq.sdk.json import converter
+from bunq.sdk.model import core
+from bunq.sdk.model.device_server_internal import DeviceServerInternal
 from bunq.sdk.model.generated import endpoint
 
 
@@ -119,13 +121,11 @@ def _register_device(self, device_description,
         :rtype: None
         """
 
-        endpoint.DeviceServer.create(
-            self,
-            {
-                endpoint.DeviceServer.FIELD_DESCRIPTION: device_description,
-                endpoint.DeviceServer.FIELD_SECRET: self.api_key,
-                endpoint.DeviceServer.FIELD_PERMITTED_IPS: permitted_ips,
-            }
+        DeviceServerInternal.create(
+            device_description,
+            self.api_key,
+            permitted_ips,
+            api_context=self
         )
 
     def _initialize_session(self):
@@ -136,8 +136,9 @@ def _initialize_session(self):
         session_server = core.SessionServer.create(self).value
         token = session_server.token.token
         expiry_time = self._get_expiry_timestamp(session_server)
+        user_id = session_server.get_referenced_object().id_
 
-        self._session_context = SessionContext(token, expiry_time)
+        self._session_context = SessionContext(token, expiry_time, user_id)
 
     @classmethod
     def _get_expiry_timestamp(cls, session_server):
@@ -377,16 +378,18 @@ class SessionContext(object):
     """
     :type _token: str
     :type _expiry_time: datetime.datetime
+    :type _user_id: int
     """
 
-    def __init__(self, token, expiry_time):
+    def __init__(self, token, expiry_time, user_id):
         """
         :type token: str
         :type expiry_time: datetime.datetime
         """
 
         self._token = token
         self._expiry_time = expiry_time
+        self._user_id = user_id
 
     @property
     def token(self):
@@ -403,3 +406,146 @@ def expiry_time(self):
         """
 
         return self._expiry_time
+
+    @property
+    def user_id(self):
+        return self._user_id
+
+
+class UserContext(object):
+    _ERROR_UNEXPECTED_USER_INSTANCE = '"{}" is unexpected user instance.'
+    _ERROR_NO_ACTIVE_MONETARY_ACCOUNT_FOUND = \
+        'No active monetary account found.'
+    _STATUS_ACTIVE = 'ACTIVE'
+
+    def __init__(self, user_id):
+        """
+        :type user_id: int
+        """
+
+        self._user_id = user_id
+        self._user_person = None
+        self._user_company = None
+        self._primary_monetary_account = None
+
+        user_object = endpoint.User.list().value[0].get_referenced_object()
+        self._set_user(user_object)
+
+    def _set_user(self, user):
+        if isinstance(user, endpoint.UserPerson):
+            self._user_person = user
+
+        elif isinstance(user, endpoint.UserCompany):
+            self._user_company = user
+
+        else:
+            raise BunqException(
+                self._ERROR_UNEXPECTED_USER_INSTANCE.format(user.__class__))
+
+    def init_main_monetary_account(self):
+        all_monetary_account = endpoint.MonetaryAccountBank.list().value
+
+        for account in all_monetary_account:
+            if account.status == self._STATUS_ACTIVE:
+                self._primary_monetary_account = account
+
+                return
+
+        raise BunqException(self._ERROR_NO_ACTIVE_MONETARY_ACCOUNT_FOUND)
+
+    @property
+    def user_id(self):
+        return self._user_id
+
+    def is_only_user_person_set(self):
+        """
+        :rtype: bool
+        """
+
+        return self._user_person is not None and self._user_company is None
+
+    def is_only_user_company_set(self):
+        """
+        :rtype: bool
+        """
+
+        return self._user_company is not None and self._user_person is None
+
+    def is_both_user_type_set(self):
+        """
+        :rtype: bool
+        """
+
+        return self._user_company is not None and self._user_person is not None
+
+    @property
+    def user_company(self):
+        """
+        :rtype: endpoint.UserCompany
+        """
+
+        return self._user_company
+
+    @property
+    def user_person(self):
+        """
+        :rtype: endpoint.UserPerson
+        """
+
+        return self._user_person
+
+    @property
+    def primary_monetary_account(self):
+        """
+        :rtype: endpoint.MonetaryAccountBank
+        """
+
+        return self._primary_monetary_account
+
+
+class BunqContext(object):
+    _ERROR_CLASS_SHOULD_NOT_BE_INITIALIZED = \
+        'This class should not be instantiated'
+    _ERROR_API_CONTEXT_HAS_NOT_BEEN_LOADED = \
+        'ApiContext has not been loaded. Please load ApiContext in BunqContext'
+    _ERROR_USER_CONTEXT_HAS_NOT_BEEN_LOADED = \
+        'UserContext has not been loaded, please load this' \
+        ' by loading ApiContext.'
+
+    _api_context = None
+    _user_context = None
+
+    def __init__(self):
+        raise TypeError(self._ERROR_CLASS_SHOULD_NOT_BE_INITIALIZED)
+
+    @classmethod
+    def load_api_context(cls, api_context):
+        """
+        :type api_context: ApiContext
+        """
+
+        cls._api_context = api_context
+        cls._user_context = UserContext(api_context.session_context.user_id)
+        cls._user_context.init_main_monetary_account()
+
+    @classmethod
+    def api_context(cls):
+        """
+        :rtype: ApiContext
+        """
+
+        if cls._api_context is not None:
+            return cls._api_context
+
+        raise BunqException(cls._ERROR_API_CONTEXT_HAS_NOT_BEEN_LOADED)
+
+    @classmethod
+    def user_context(cls):
+        """
+        :rtype: UserContext
+        """
+
+        if cls._user_context is not None:
+            return cls._user_context
+
+        raise BunqException(cls._ERROR_USER_CONTEXT_HAS_NOT_BEEN_LOADED)

bunq/sdk/model/core.py

@@ -1,5 +1,7 @@
 from bunq.sdk import client
 from bunq.sdk.json import converter
+from bunq.sdk.exception import BunqException
+from bunq.sdk import context
 
 
 class AnchoredObjectInterface:
@@ -135,6 +137,35 @@ def _from_json_list(cls, response_raw, wrapper=None):
         return client.BunqResponse(array_deserialized, response_raw.headers,
                                    pagination)
 
+    @classmethod
+    def _get_api_context(cls):
+        """
+        :rtype: context.ApiContext
+        """
+
+        return context.BunqContext.api_context()
+
+    @classmethod
+    def _determine_user_id(cls):
+        """
+        :rtype: int
+        """
+
+        return context.BunqContext.user_context().user_id
+
+    @classmethod
+    def _determine_monetary_account_id(cls, monetary_account_id=None):
+        """
+        :type monetary_account_id: int
+
+        :rtype: int
+        """
+
+        if monetary_account_id is None:
+            return context.BunqContext.user_context().primary_monetary_account.id_
+
+        return monetary_account_id
+
 
 class Id(BunqModel):
     """
@@ -366,6 +397,9 @@ class SessionServer(BunqModel):
     # Field constants
     FIELD_SECRET = "secret"
 
+    # Error constants
+    _ERROR_ALL_FIELD_IS_NULL = 'All fields are null'
+
     def __init__(self):
         self._id_ = None
         self._token = None
@@ -442,3 +476,16 @@ def is_all_field_none(self):
             return False
 
         return True
+
+    def get_referenced_object(self):
+        """
+        :rtype: BunqModel
+        """
+
+        if self._user_person is not None:
+            return self._user_person
+
+        if self._user_company is not None:
+            return self._user_company
+
+        raise BunqException(self._ERROR_ALL_FIELD_IS_NULL)

bunq/sdk/model/device_server_internal.py

@@ -0,0 +1,60 @@
+from bunq.sdk.model.generated.endpoint import DeviceServer
+from bunq.sdk.model.generated.endpoint import BunqResponseInt
+from bunq.sdk import client
+from bunq.sdk.json import converter
+from bunq.sdk.exception import BunqException
+
+
+class DeviceServerInternal(DeviceServer):
+    _ERROR_API_CONTEXT_IS_NULL = 'ApiContext should not be None,' \
+                                 ' use the generated class instead.'
+
+    @classmethod
+    def create(cls, description, secret, permitted_ips=None,
+               custom_headers=None, api_context=None):
+        """
+        Create a new DeviceServer providing the installation token in the header
+        and signing the request with the private part of the key you used to
+        create the installation. The API Key that you are using will be bound to
+        the IP address of the DeviceServer which you have
+        created.<br/><br/>Using a Wildcard API Key gives you the freedom to make
+        API calls even if the IP address has changed after the POST
+        device-server.<br/><br/>Find out more at this link <a
+        href="https://bunq.com/en/apikey-dynamic-ip"
+        target="_blank">https://bunq.com/en/apikey-dynamic-ip</a>.
+
+        :param description: The description of the DeviceServer. This is only
+        for your own reference when reading the DeviceServer again.
+        :type description: str
+        :param secret: The API key. You can request an API key in the bunq app.
+        :type secret: str
+        :param permitted_ips: An array of IPs (v4 or v6) this DeviceServer will
+        be able to do calls from. These will be linked to the API key.
+        :type permitted_ips: list[str]
+        :type custom_headers: dict[str, str]|None
+        :type api_context: context.ApiContext
+
+        :rtype: BunqResponseInt
+        """
+
+        if api_context is None:
+            raise BunqException(cls._ERROR_API_CONTEXT_IS_NULL)
+
+        if custom_headers is None:
+            custom_headers = {}
+
+        request_map = {
+            cls.FIELD_DESCRIPTION: description,
+            cls.FIELD_SECRET: secret,
+            cls.FIELD_PERMITTED_IPS: permitted_ips
+        }
+
+        api_client = client.ApiClient(api_context)
+        request_bytes = converter.class_to_json(request_map).encode()
+        endpoint_url = cls._ENDPOINT_URL_CREATE
+        response_raw = api_client.post(endpoint_url, request_bytes,
+                                       custom_headers)
+
+        return BunqResponseInt.cast_from_bunq_response(
+            cls._process_for_id(response_raw)
+        )