From 2fb8645d35aafa8bb94212ca1d2233d8dd6e3b31 Mon Sep 17 00:00:00 2001 From: Angelo Melonas Date: Wed, 18 Sep 2019 15:01:51 +0200 Subject: [PATCH] Refactored the json/adapters. --- bunq/__init__.py | 34 +- bunq/sdk/http/api_client.py | 4 +- bunq/sdk/json/adapters.py | 648 ------------------ .../sdk/json/anchored_object_model_adapter.py | 70 ++ bunq/sdk/json/api_environment_type_adapter.py | 27 + bunq/sdk/json/date_time_adapter.py | 31 + bunq/sdk/json/float_adapter.py | 30 + bunq/sdk/json/geolocation_adapter.py | 61 ++ bunq/sdk/json/installation_adapter.py | 67 ++ bunq/sdk/json/installation_context_adapter.py | 67 ++ .../monetary_account_reference_adapter.py | 32 + bunq/sdk/json/pagination_adapter.py | 108 +++ bunq/sdk/json/session_server_adapter.py | 108 +++ bunq/sdk/json/share_detail_adapter.py | 71 ++ 14 files changed, 696 insertions(+), 662 deletions(-) delete mode 100644 bunq/sdk/json/adapters.py create mode 100644 bunq/sdk/json/anchored_object_model_adapter.py create mode 100644 bunq/sdk/json/api_environment_type_adapter.py create mode 100644 bunq/sdk/json/date_time_adapter.py create mode 100644 bunq/sdk/json/float_adapter.py create mode 100644 bunq/sdk/json/geolocation_adapter.py create mode 100644 bunq/sdk/json/installation_adapter.py create mode 100644 bunq/sdk/json/installation_context_adapter.py create mode 100644 bunq/sdk/json/monetary_account_reference_adapter.py create mode 100644 bunq/sdk/json/pagination_adapter.py create mode 100644 bunq/sdk/json/session_server_adapter.py create mode 100644 bunq/sdk/json/share_detail_adapter.py diff --git a/bunq/__init__.py b/bunq/__init__.py index b91cd8f..cdd9319 100644 --- a/bunq/__init__.py +++ b/bunq/__init__.py @@ -13,41 +13,51 @@ def initialize_converter(): from bunq.sdk.http import api_client from bunq.sdk.context import api_context - from bunq.sdk.json import adapters from bunq.sdk.json import converter from bunq.sdk.model.generated import object_ from bunq.sdk.model.generated import endpoint from bunq.sdk.model.core.installation import Installation from bunq.sdk.model.core.session_server import SessionServer + from bunq.sdk.json.installation_adapter import InstallationAdapter + from bunq.sdk.json.session_server_adapter import SessionServerAdapter + from bunq.sdk.json.installation_context_adapter import InstallationContextAdapter + from bunq.sdk.json.api_environment_type_adapter import ApiEnvironmentTypeAdapter + from bunq.sdk.json.float_adapter import FloatAdapter + from bunq.sdk.json.geolocation_adapter import GeolocationAdapter + from bunq.sdk.json.monetary_account_reference_adapter import MonetaryAccountReferenceAdapter + from bunq.sdk.json.share_detail_adapter import ShareDetailAdapter + from bunq.sdk.json.date_time_adapter import DateTimeAdapter + from bunq.sdk.json.pagination_adapter import PaginationAdapter - converter.register_adapter(Installation, adapters.InstallationAdapter) + converter.register_adapter(Installation, InstallationAdapter) converter.register_adapter( SessionServer, - adapters.SessionServerAdapter + SessionServerAdapter ) converter.register_adapter( api_context.InstallationContext, - adapters.InstallationContextAdapter + InstallationContextAdapter ) converter.register_adapter( api_context.ApiEnvironmentType, - adapters.ApiEnvironmentTypeAdapter + ApiEnvironmentTypeAdapter ) - converter.register_adapter(float, adapters.FloatAdapter) - converter.register_adapter(object_.Geolocation, adapters.GeolocationAdapter) + converter.register_adapter(float, FloatAdapter) + converter.register_adapter(object_.Geolocation, GeolocationAdapter) converter.register_adapter( object_.MonetaryAccountReference, - adapters.MonetaryAccountReferenceAdapter + MonetaryAccountReferenceAdapter ) - converter.register_adapter(object_.ShareDetail, adapters.ShareDetailAdapter) - converter.register_adapter(datetime.datetime, adapters.DateTimeAdapter) - converter.register_adapter(Pagination, adapters.PaginationAdapter) + converter.register_adapter(object_.ShareDetail, ShareDetailAdapter) + converter.register_adapter(datetime.datetime, DateTimeAdapter) + converter.register_adapter(Pagination, PaginationAdapter) def register_anchor_adapter(class_to_register): if issubclass(class_to_register, AnchoredObjectInterface): + from bunq.sdk.json.anchored_object_model_adapter import AnchoredObjectModelAdapter converter.register_adapter( class_to_register, - adapters.AnchoredObjectModelAdapter + AnchoredObjectModelAdapter ) def get_class(class_string_to_get): diff --git a/bunq/sdk/http/api_client.py b/bunq/sdk/http/api_client.py index 62f311d..889d463 100644 --- a/bunq/sdk/http/api_client.py +++ b/bunq/sdk/http/api_client.py @@ -285,7 +285,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 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): """ @@ -335,4 +335,4 @@ def delete(self, uri_relative, custom_headers): self._BYTES_EMPTY, {}, custom_headers - ) \ No newline at end of file + ) diff --git a/bunq/sdk/json/adapters.py b/bunq/sdk/json/adapters.py deleted file mode 100644 index 894d269..0000000 --- a/bunq/sdk/json/adapters.py +++ /dev/null @@ -1,648 +0,0 @@ -import datetime -import urllib.parse as urlparse - -from bunq import Pagination, AnchoredObjectInterface -from bunq.sdk.context import api_context -from bunq.sdk.exception.bunq_exception import BunqException -from bunq.sdk.model.core.id import Id -from bunq.sdk.model.core.public_key_server import PublicKeyServer -from bunq.sdk.model.core.session_token import SessionToken -from bunq.sdk.security import security -from bunq.sdk.json import converter -from bunq.sdk.model.generated import endpoint -from bunq.sdk.model.generated import object_ - - -class AnchoredObjectModelAdapter(converter.JsonAdapter): - _ERROR_MODEL_NOT_FOUND = '{} is not in endpoint nor object.' - - __STRING_FORMAT_UNDERSCORE = '_' - - _override_field_map = { - 'ScheduledPayment': 'SchedulePayment', - 'ScheduledInstance': 'ScheduleInstance', - } - - @classmethod - def deserialize(cls, cls_target, obj_raw): - """ - :type cls_target: BunqModel - :type obj_raw: int|str|bool|float|list|dict|None - - :rtype: T - """ - - model_ = super()._deserialize_default(cls_target, obj_raw) - - if isinstance( - model_, - AnchoredObjectInterface - ) and model_.is_all_field_none(): - for field in model_.__dict__: - object_class = cls._get_object_class(field) - contents = super()._deserialize_default(object_class, obj_raw) - - if contents.is_all_field_none(): - setattr(model_, field, None) - else: - setattr(model_, field, contents) - - return model_ - - @classmethod - def can_serialize(cls): - return False - - @classmethod - def _get_object_class(cls, class_name): - """ - :type class_name: str - :rtype: BunqModel - """ - - class_name = class_name.lstrip(cls.__STRING_FORMAT_UNDERSCORE) - - if class_name in cls._override_field_map: - class_name = cls._override_field_map[class_name] - - try: - return getattr(endpoint, class_name) - except AttributeError: - pass - - try: - return getattr(object_, class_name) - except AttributeError: - pass - - raise BunqException(cls._ERROR_MODEL_NOT_FOUND.format(class_name)) - - -class InstallationAdapter(converter.JsonAdapter): - # Id constants - _ATTRIBUTE_ID = '_id_' - _INDEX_ID = 0 - _FIELD_ID = 'Id' - - # Token constants - _ATTRIBUTE_TOKEN = '_token' - _INDEX_TOKEN = 1 - _FIELD_TOKEN = 'Token' - - # Server Public Key constants - _ATTRIBUTE_SERVER_PUBLIC_KEY = '_server_public_key' - _INDEX_SERVER_PUBLIC_KEY = 2 - _FIELD_SERVER_PUBLIC_KEY = 'ServerPublicKey' - - @classmethod - def deserialize(cls, target_class, array): - """ - :type target_class: Installation|type - :type array: list - - :rtype: Installation - """ - - installation = target_class.__new__(target_class) - server_public_key_wrapped = array[cls._INDEX_SERVER_PUBLIC_KEY] - installation.__dict__ = { - cls._ATTRIBUTE_ID: converter.deserialize( - Id, - array[cls._INDEX_ID][cls._FIELD_ID] - ), - cls._ATTRIBUTE_TOKEN: converter.deserialize( - SessionToken, - array[cls._INDEX_TOKEN][cls._FIELD_TOKEN] - ), - cls._ATTRIBUTE_SERVER_PUBLIC_KEY: converter.deserialize( - PublicKeyServer, - server_public_key_wrapped[cls._FIELD_SERVER_PUBLIC_KEY] - ), - } - - return installation - - @classmethod - def serialize(cls, installation): - """ - :type installation: Installation - - :rtype: list - """ - - return [ - {cls._FIELD_ID: converter.serialize(installation.id_)}, - {cls._FIELD_TOKEN: converter.serialize(installation.token)}, - { - cls._FIELD_SERVER_PUBLIC_KEY: converter.serialize( - installation.server_public_key - ), - }, - ] - - -class SessionServerAdapter(converter.JsonAdapter): - # Error constants. - _ERROR_COULD_NOT_DETERMINE_USER = 'Could not determine user.' - - # Id constants - _ATTRIBUTE_ID = '_id_' - _INDEX_ID = 0 - _FIELD_ID = 'Id' - - # Token constants - _ATTRIBUTE_TOKEN = '_token' - _INDEX_TOKEN = 1 - _FIELD_TOKEN = 'Token' - - # User constants - _INDEX_USER = 2 - - # UserCompany constants - _ATTRIBUTE_USER_COMPANY = '_user_company' - _FIELD_USER_COMPANY = 'UserCompany' - - # UserPerson constants - _ATTRIBUTE_USER_PERSON = '_user_person' - _FIELD_USER_PERSON = 'UserPerson' - - # UserApiKey constants - _ATTRIBUTE_USER_API_KEY = '_user_api_key' - _FIELD_USER_API_KEY = 'UserApiKey' - - @classmethod - def deserialize(cls, target_class, array): - """ - :type target_class: SessionServer|type - :type array: list - - :rtype: SessionServer - """ - - session_server = target_class.__new__(target_class) - session_server.__dict__ = { - cls._ATTRIBUTE_ID: converter.deserialize( - Id, - array[cls._INDEX_ID][cls._FIELD_ID] - ), - cls._ATTRIBUTE_TOKEN: converter.deserialize( - SessionToken, - array[cls._INDEX_TOKEN][cls._FIELD_TOKEN] - ), - cls._ATTRIBUTE_USER_COMPANY: None, - cls._ATTRIBUTE_USER_PERSON: None, - } - - user_dict_wrapped = array[cls._INDEX_USER] - - if cls._FIELD_USER_COMPANY in user_dict_wrapped: - session_server.__dict__[cls._ATTRIBUTE_USER_COMPANY] = \ - converter.deserialize( - endpoint.UserCompany, - user_dict_wrapped[cls._FIELD_USER_COMPANY] - ) - elif cls._FIELD_USER_PERSON in user_dict_wrapped: - session_server.__dict__[cls._ATTRIBUTE_USER_PERSON] = \ - converter.deserialize( - endpoint.UserPerson, - user_dict_wrapped[cls._FIELD_USER_PERSON] - ) - elif cls._FIELD_USER_API_KEY in user_dict_wrapped: - session_server.__dict__[cls._ATTRIBUTE_USER_API_KEY] = \ - converter.deserialize( - endpoint.UserApiKey, - user_dict_wrapped[cls._FIELD_USER_API_KEY] - ) - else: - raise BunqException(cls._ERROR_COULD_NOT_DETERMINE_USER) - - return session_server - - @classmethod - def serialize(cls, session_server): - """ - :type session_server: SessionServer - - :rtype: list - """ - - return [ - {cls._FIELD_ID: converter.serialize(session_server.id_)}, - {cls._FIELD_TOKEN: converter.serialize(session_server.token)}, - { - cls._FIELD_USER_COMPANY: - converter.serialize(session_server.user_company), - }, - { - cls._FIELD_USER_PERSON: - converter.serialize(session_server.user_person), - }, - { - cls._FIELD_USER_API_KEY: - converter.serialize(session_server.user_api_key), - }, - ] - - -class InstallationContextAdapter(converter.JsonAdapter): - # Attribute/Field constants - _ATTRIBUTE_TOKEN = '_token' - _FIELD_TOKEN = 'token' - - _ATTRIBUTE_PRIVATE_KEY_CLIENT = '_private_key_client' - _FIELD_PRIVATE_KEY_CLIENT = 'private_key_client' - - _ATTRIBUTE_PUBLIC_KEY_CLIENT = '_public_key_client' - _FIELD_PUBLIC_KEY_CLIENT = 'public_key_client' - - _ATTRIBUTE_PUBLIC_KEY_SERVER = '_public_key_server' - _FIELD_PUBLIC_KEY_SERVER = 'public_key_server' - - @classmethod - def deserialize(cls, target_class, obj): - """ - :type target_class: api_context.InstallationContext|type - :type obj: dict - - :rtype: context.InstallationContext - """ - - installation_context = target_class.__new__(target_class) - private_key_client = security.rsa_key_from_string( - obj[cls._FIELD_PRIVATE_KEY_CLIENT] - ) - public_key_client = security.rsa_key_from_string( - obj[cls._FIELD_PUBLIC_KEY_CLIENT] - ) - public_key_server = security.rsa_key_from_string( - obj[cls._FIELD_PUBLIC_KEY_SERVER] - ) - installation_context.__dict__ = { - cls._ATTRIBUTE_TOKEN: obj[cls._FIELD_TOKEN], - cls._ATTRIBUTE_PRIVATE_KEY_CLIENT: private_key_client, - cls._ATTRIBUTE_PUBLIC_KEY_CLIENT: public_key_client, - cls._ATTRIBUTE_PUBLIC_KEY_SERVER: public_key_server, - } - - return installation_context - - @classmethod - def serialize(cls, installation_context): - """ - :type installation_context: api_context.InstallationContext - - :rtype: dict - """ - - return { - cls._FIELD_TOKEN: installation_context.token, - cls._FIELD_PUBLIC_KEY_CLIENT: security.public_key_to_string( - installation_context.private_key_client.publickey() - ), - cls._FIELD_PRIVATE_KEY_CLIENT: security.private_key_to_string( - installation_context.private_key_client - ), - cls._FIELD_PUBLIC_KEY_SERVER: security.public_key_to_string( - installation_context.public_key_server - ), - } - - -class ApiEnvironmentTypeAdapter(converter.JsonAdapter): - @classmethod - def deserialize(cls, target_class, name): - """ - :type target_class: api_context.ApiEnvironmentType|type - :type name: str - - :rtype: context.ApiEnvironmentType - """ - - _ = target_class - - return api_context.ApiEnvironmentType[name] - - @classmethod - def serialize(cls, api_environment_type): - """ - :type api_environment_type: api_context.ApiEnvironmentType - - :rtype: str - """ - - return api_environment_type.name - - -class FloatAdapter(converter.JsonAdapter): - # Precision to round the floats to before outputting them - _PRECISION_FLOAT = 2 - - @classmethod - def deserialize(cls, target_class, string): - """ - :type target_class: float|type - :type string: str - - :rtype: float - """ - - _ = target_class - - return float(string) - - @classmethod - def serialize(cls, number): - """ - :type number: float - - :rtype: str - """ - - return str(round(number, cls._PRECISION_FLOAT)) - - -class GeolocationAdapter(converter.JsonAdapter): - # Field constants - _FIELD_LATITUDE = 'latitude' - _FIELD_LONGITUDE = 'longitude' - _FIELD_ALTITUDE = 'altitude' - _FIELD_RADIUS = 'radius' - - @classmethod - def can_deserialize(cls): - """ - :rtype: bool - """ - - return False - - @classmethod - def deserialize(cls, target_class, obj): - """ - :type target_class: float|type - :type obj: dict - - :raise: NotImplementedError - """ - - _ = target_class, obj - - raise NotImplementedError() - - @classmethod - def serialize(cls, geolocation): - """ - :type geolocation: object_.Geolocation - - :rtype: dict - """ - - obj = {} - - cls.add_if_not_none(obj, cls._FIELD_LATITUDE, geolocation.latitude) - cls.add_if_not_none(obj, cls._FIELD_LONGITUDE, geolocation.longitude) - cls.add_if_not_none(obj, cls._FIELD_ALTITUDE, geolocation.altitude) - cls.add_if_not_none(obj, cls._FIELD_RADIUS, geolocation.radius) - - return obj - - @classmethod - def add_if_not_none(cls, dict_, key, value): - """ - :type dict_: dict[str, str] - :type key: str - :type value: float - - :rtype: None - """ - - if value is not None: - dict_[key] = str(value) - - -class MonetaryAccountReferenceAdapter(converter.JsonAdapter): - @classmethod - def deserialize(cls, target_class, obj): - """ - :type target_class: object_.MonetaryAccountReference|type - :type obj: dict - - :rtype: object_.MonetaryAccountReference - """ - - label_monetary_account = converter.deserialize( - object_.LabelMonetaryAccount, - obj - ) - - return target_class.create_from_label_monetary_account( - label_monetary_account - ) - - @classmethod - def serialize(cls, monetary_account_reference): - """ - :type monetary_account_reference: object_.MonetaryAccountReference - - :rtype: dict - """ - - return converter.serialize(monetary_account_reference.pointer) - - -class ShareDetailAdapter(converter.JsonAdapter): - # Attribute/Field constants - _ATTRIBUTE_PAYMENT = 'payment' - _FIELD_PAYMENT = 'ShareDetailPayment' - - _ATTRIBUTE_READ_ONLY = 'read_only' - _FIELD_READ_ONLY = 'ShareDetailReadOnly' - - _ATTRIBUTE_DRAFT_PAYMENT = 'draft_payment' - _FIELD_DRAFT_PAYMENT = 'ShareDetailDraftPayment' - - @classmethod - def deserialize(cls, target_class, obj): - """ - :type target_class: object_.ShareDetail|type - :type obj: dict - - :rtype: object_.ShareDetail - """ - - share_detail = target_class.__new__(target_class) - share_detail.__dict__ = { - cls._ATTRIBUTE_PAYMENT: converter.deserialize( - object_.ShareDetailPayment, - cls._get_field_or_none(cls._FIELD_DRAFT_PAYMENT, obj) - ), - cls._ATTRIBUTE_READ_ONLY: converter.deserialize( - object_.ShareDetailReadOnly, - cls._get_field_or_none(cls._FIELD_READ_ONLY, obj) - ), - cls._ATTRIBUTE_DRAFT_PAYMENT: converter.deserialize( - object_.ShareDetailDraftPayment, - cls._get_field_or_none(cls._FIELD_DRAFT_PAYMENT, obj) - ), - } - - return share_detail - - @staticmethod - def _get_field_or_none(field, obj): - """ - :type field: str - :type obj: dict - - :return: dict|None - """ - - return obj[field] if field in obj else None - - @classmethod - def serialize(cls, share_detail): - """ - :type share_detail: object_.ShareDetail - - :rtype: dict - """ - - return { - cls._FIELD_PAYMENT: converter.serialize( - share_detail._payment_field_for_request), - cls._FIELD_READ_ONLY: converter.serialize( - share_detail._read_only_field_for_request), - cls._FIELD_DRAFT_PAYMENT: converter.serialize( - share_detail._draft_payment - ), - } - - -class DateTimeAdapter(converter.JsonAdapter): - # bunq timestamp format - _FORMAT_TIMESTAMP = '%Y-%m-%d %H:%M:%S.%f' - - @classmethod - def deserialize(cls, target_class, string): - """ - :type target_class: datetime.datetime|type - :type string: str - - :rtype: datetime.datetime - """ - - return target_class.strptime(string, cls._FORMAT_TIMESTAMP) - - @classmethod - def serialize(cls, timestamp): - """ - :type timestamp: datetime.datetime - - :rtype: dict - """ - - return timestamp.strftime(cls._FORMAT_TIMESTAMP) - - -class PaginationAdapter(converter.JsonAdapter): - # Raw pagination response field constants. - _FIELD_FUTURE_URL = 'future_url' - _FIELD_NEWER_URL = 'newer_url' - _FIELD_OLDER_URL = 'older_url' - - # Processed pagination field constants. - _FIELD_OLDER_ID = 'older_id' - _FIELD_NEWER_ID = 'newer_id' - _FIELD_FUTURE_ID = 'future_id' - _FIELD_COUNT = 'count' - - # Very first index in an array. - _INDEX_FIRST = 0 - - @classmethod - def deserialize(cls, target_class, pagination_response): - """ - :type target_class: Pagination|type - :type pagination_response: dict - - :rtype: client.Pagination - """ - - pagination = Pagination() - pagination.__dict__.update( - cls.parse_pagination_dict(pagination_response) - ) - - return pagination - - @classmethod - def parse_pagination_dict(cls, response_obj): - """ - :type response_obj: dict - - :rtype: dict - """ - - pagination_dict = {} - - cls.update_dict_id_field_from_response_field( - pagination_dict, - cls._FIELD_OLDER_ID, - response_obj, - cls._FIELD_OLDER_URL, - Pagination.PARAM_OLDER_ID - ) - cls.update_dict_id_field_from_response_field( - pagination_dict, - cls._FIELD_NEWER_ID, - response_obj, - cls._FIELD_NEWER_URL, - Pagination.PARAM_NEWER_ID - ) - cls.update_dict_id_field_from_response_field( - pagination_dict, - cls._FIELD_FUTURE_ID, - response_obj, - cls._FIELD_FUTURE_URL, - Pagination.PARAM_NEWER_ID - ) - - return pagination_dict - - @classmethod - def update_dict_id_field_from_response_field(cls, dict_, dict_id_field, - response_obj, response_field, - response_param): - """ - :type dict_: dict - :type dict_id_field: str - :type response_obj: dict - :type response_field: str - :type response_param: str - """ - - url = response_obj[response_field] - - if url is not None: - url_parsed = urlparse.urlparse(url) - parameters = urlparse.parse_qs(url_parsed.query) - dict_[dict_id_field] = int( - parameters[response_param][cls._INDEX_FIRST] - ) - - if cls._FIELD_COUNT in parameters and cls._FIELD_COUNT not in dict_: - dict_[cls._FIELD_COUNT] = int( - parameters[Pagination.PARAM_COUNT][cls._INDEX_FIRST] - ) - - @classmethod - def serialize(cls, pagination): - """ - :type pagination: Pagination - - :raise: NotImplementedError - """ - - _ = pagination - - raise NotImplementedError() diff --git a/bunq/sdk/json/anchored_object_model_adapter.py b/bunq/sdk/json/anchored_object_model_adapter.py new file mode 100644 index 0000000..27203b2 --- /dev/null +++ b/bunq/sdk/json/anchored_object_model_adapter.py @@ -0,0 +1,70 @@ +from bunq import AnchoredObjectInterface +from bunq.sdk.exception.bunq_exception import BunqException +from bunq.sdk.json import converter +from bunq.sdk.model.generated import endpoint +from bunq.sdk.model.generated import object_ + + +class AnchoredObjectModelAdapter(converter.JsonAdapter): + _ERROR_MODEL_NOT_FOUND = '{} is not in endpoint nor object.' + + __STRING_FORMAT_UNDERSCORE = '_' + + _override_field_map = { + 'ScheduledPayment': 'SchedulePayment', + 'ScheduledInstance': 'ScheduleInstance', + } + + @classmethod + def deserialize(cls, cls_target, obj_raw): + """ + :type cls_target: BunqModel + :type obj_raw: int|str|bool|float|list|dict|None + + :rtype: T + """ + + model_ = super()._deserialize_default(cls_target, obj_raw) + + if isinstance( + model_, + AnchoredObjectInterface + ) and model_.is_all_field_none(): + for field in model_.__dict__: + object_class = cls._get_object_class(field) + contents = super()._deserialize_default(object_class, obj_raw) + + if contents.is_all_field_none(): + setattr(model_, field, None) + else: + setattr(model_, field, contents) + + return model_ + + @classmethod + def can_serialize(cls): + return False + + @classmethod + def _get_object_class(cls, class_name): + """ + :type class_name: str + :rtype: BunqModel + """ + + class_name = class_name.lstrip(cls.__STRING_FORMAT_UNDERSCORE) + + if class_name in cls._override_field_map: + class_name = cls._override_field_map[class_name] + + try: + return getattr(endpoint, class_name) + except AttributeError: + pass + + try: + return getattr(object_, class_name) + except AttributeError: + pass + + raise BunqException(cls._ERROR_MODEL_NOT_FOUND.format(class_name)) diff --git a/bunq/sdk/json/api_environment_type_adapter.py b/bunq/sdk/json/api_environment_type_adapter.py new file mode 100644 index 0000000..e01fde0 --- /dev/null +++ b/bunq/sdk/json/api_environment_type_adapter.py @@ -0,0 +1,27 @@ +from bunq.sdk.context import api_context +from bunq.sdk.json import converter + + +class ApiEnvironmentTypeAdapter(converter.JsonAdapter): + @classmethod + def deserialize(cls, target_class, name): + """ + :type target_class: api_context.ApiEnvironmentType|type + :type name: str + + :rtype: context.ApiEnvironmentType + """ + + _ = target_class + + return api_context.ApiEnvironmentType[name] + + @classmethod + def serialize(cls, api_environment_type): + """ + :type api_environment_type: api_context.ApiEnvironmentType + + :rtype: str + """ + + return api_environment_type.name diff --git a/bunq/sdk/json/date_time_adapter.py b/bunq/sdk/json/date_time_adapter.py new file mode 100644 index 0000000..a54a3d9 --- /dev/null +++ b/bunq/sdk/json/date_time_adapter.py @@ -0,0 +1,31 @@ +import datetime + +from bunq.sdk.json import converter + + +class DateTimeAdapter(converter.JsonAdapter): + # bunq timestamp format + _FORMAT_TIMESTAMP = '%Y-%m-%d %H:%M:%S.%f' + + @classmethod + def deserialize(cls, target_class, string): + """ + :type target_class: datetime.datetime|type + :type string: str + + :rtype: datetime.datetime + """ + + return target_class.strptime(string, cls._FORMAT_TIMESTAMP) + + @classmethod + def serialize(cls, timestamp): + """ + :type timestamp: datetime.datetime + + :rtype: dict + """ + + return timestamp.strftime(cls._FORMAT_TIMESTAMP) + + diff --git a/bunq/sdk/json/float_adapter.py b/bunq/sdk/json/float_adapter.py new file mode 100644 index 0000000..7e7fbc3 --- /dev/null +++ b/bunq/sdk/json/float_adapter.py @@ -0,0 +1,30 @@ +from bunq.sdk.json import converter + + +class FloatAdapter(converter.JsonAdapter): + # Precision to round the floats to before outputting them + _PRECISION_FLOAT = 2 + + @classmethod + def deserialize(cls, target_class, string): + """ + :type target_class: float|type + :type string: str + + :rtype: float + """ + + _ = target_class + + return float(string) + + @classmethod + def serialize(cls, number): + """ + :type number: float + + :rtype: str + """ + + return str(round(number, cls._PRECISION_FLOAT)) + diff --git a/bunq/sdk/json/geolocation_adapter.py b/bunq/sdk/json/geolocation_adapter.py new file mode 100644 index 0000000..38e66fb --- /dev/null +++ b/bunq/sdk/json/geolocation_adapter.py @@ -0,0 +1,61 @@ +from bunq.sdk.json import converter +from bunq.sdk.model.generated import object_ + + +class GeolocationAdapter(converter.JsonAdapter): + # Field constants + _FIELD_LATITUDE = 'latitude' + _FIELD_LONGITUDE = 'longitude' + _FIELD_ALTITUDE = 'altitude' + _FIELD_RADIUS = 'radius' + + @classmethod + def can_deserialize(cls): + """ + :rtype: bool + """ + + return False + + @classmethod + def deserialize(cls, target_class, obj): + """ + :type target_class: float|type + :type obj: dict + + :raise: NotImplementedError + """ + + _ = target_class, obj + + raise NotImplementedError() + + @classmethod + def serialize(cls, geolocation): + """ + :type geolocation: object_.Geolocation + + :rtype: dict + """ + + obj = {} + + cls.add_if_not_none(obj, cls._FIELD_LATITUDE, geolocation.latitude) + cls.add_if_not_none(obj, cls._FIELD_LONGITUDE, geolocation.longitude) + cls.add_if_not_none(obj, cls._FIELD_ALTITUDE, geolocation.altitude) + cls.add_if_not_none(obj, cls._FIELD_RADIUS, geolocation.radius) + + return obj + + @classmethod + def add_if_not_none(cls, dict_, key, value): + """ + :type dict_: dict[str, str] + :type key: str + :type value: float + + :rtype: None + """ + + if value is not None: + dict_[key] = str(value) diff --git a/bunq/sdk/json/installation_adapter.py b/bunq/sdk/json/installation_adapter.py new file mode 100644 index 0000000..8f85a3d --- /dev/null +++ b/bunq/sdk/json/installation_adapter.py @@ -0,0 +1,67 @@ +from bunq.sdk.json import converter +from bunq.sdk.model.core.id import Id +from bunq.sdk.model.core.public_key_server import PublicKeyServer +from bunq.sdk.model.core.session_token import SessionToken + + +class InstallationAdapter(converter.JsonAdapter): + # Id constants + _ATTRIBUTE_ID = '_id_' + _INDEX_ID = 0 + _FIELD_ID = 'Id' + + # Token constants + _ATTRIBUTE_TOKEN = '_token' + _INDEX_TOKEN = 1 + _FIELD_TOKEN = 'Token' + + # Server Public Key constants + _ATTRIBUTE_SERVER_PUBLIC_KEY = '_server_public_key' + _INDEX_SERVER_PUBLIC_KEY = 2 + _FIELD_SERVER_PUBLIC_KEY = 'ServerPublicKey' + + @classmethod + def deserialize(cls, target_class, array): + """ + :type target_class: Installation|type + :type array: list + + :rtype: Installation + """ + + installation = target_class.__new__(target_class) + server_public_key_wrapped = array[cls._INDEX_SERVER_PUBLIC_KEY] + installation.__dict__ = { + cls._ATTRIBUTE_ID: converter.deserialize( + Id, + array[cls._INDEX_ID][cls._FIELD_ID] + ), + cls._ATTRIBUTE_TOKEN: converter.deserialize( + SessionToken, + array[cls._INDEX_TOKEN][cls._FIELD_TOKEN] + ), + cls._ATTRIBUTE_SERVER_PUBLIC_KEY: converter.deserialize( + PublicKeyServer, + server_public_key_wrapped[cls._FIELD_SERVER_PUBLIC_KEY] + ), + } + + return installation + + @classmethod + def serialize(cls, installation): + """ + :type installation: Installation + + :rtype: list + """ + + return [ + {cls._FIELD_ID: converter.serialize(installation.id_)}, + {cls._FIELD_TOKEN: converter.serialize(installation.token)}, + { + cls._FIELD_SERVER_PUBLIC_KEY: converter.serialize( + installation.server_public_key + ), + }, + ] diff --git a/bunq/sdk/json/installation_context_adapter.py b/bunq/sdk/json/installation_context_adapter.py new file mode 100644 index 0000000..5b5d164 --- /dev/null +++ b/bunq/sdk/json/installation_context_adapter.py @@ -0,0 +1,67 @@ +from bunq.sdk.context import api_context +from bunq.sdk.json import converter +from bunq.sdk.security import security + + +class InstallationContextAdapter(converter.JsonAdapter): + # Attribute/Field constants + _ATTRIBUTE_TOKEN = '_token' + _FIELD_TOKEN = 'token' + + _ATTRIBUTE_PRIVATE_KEY_CLIENT = '_private_key_client' + _FIELD_PRIVATE_KEY_CLIENT = 'private_key_client' + + _ATTRIBUTE_PUBLIC_KEY_CLIENT = '_public_key_client' + _FIELD_PUBLIC_KEY_CLIENT = 'public_key_client' + + _ATTRIBUTE_PUBLIC_KEY_SERVER = '_public_key_server' + _FIELD_PUBLIC_KEY_SERVER = 'public_key_server' + + @classmethod + def deserialize(cls, target_class, obj): + """ + :type target_class: api_context.InstallationContext|type + :type obj: dict + + :rtype: context.InstallationContext + """ + + installation_context = target_class.__new__(target_class) + private_key_client = security.rsa_key_from_string( + obj[cls._FIELD_PRIVATE_KEY_CLIENT] + ) + public_key_client = security.rsa_key_from_string( + obj[cls._FIELD_PUBLIC_KEY_CLIENT] + ) + public_key_server = security.rsa_key_from_string( + obj[cls._FIELD_PUBLIC_KEY_SERVER] + ) + installation_context.__dict__ = { + cls._ATTRIBUTE_TOKEN: obj[cls._FIELD_TOKEN], + cls._ATTRIBUTE_PRIVATE_KEY_CLIENT: private_key_client, + cls._ATTRIBUTE_PUBLIC_KEY_CLIENT: public_key_client, + cls._ATTRIBUTE_PUBLIC_KEY_SERVER: public_key_server, + } + + return installation_context + + @classmethod + def serialize(cls, installation_context): + """ + :type installation_context: api_context.InstallationContext + + :rtype: dict + """ + + return { + cls._FIELD_TOKEN: installation_context.token, + cls._FIELD_PUBLIC_KEY_CLIENT: security.public_key_to_string( + installation_context.private_key_client.publickey() + ), + cls._FIELD_PRIVATE_KEY_CLIENT: security.private_key_to_string( + installation_context.private_key_client + ), + cls._FIELD_PUBLIC_KEY_SERVER: security.public_key_to_string( + installation_context.public_key_server + ), + } diff --git a/bunq/sdk/json/monetary_account_reference_adapter.py b/bunq/sdk/json/monetary_account_reference_adapter.py new file mode 100644 index 0000000..73040d8 --- /dev/null +++ b/bunq/sdk/json/monetary_account_reference_adapter.py @@ -0,0 +1,32 @@ +from bunq.sdk.json import converter +from bunq.sdk.model.generated import object_ + + +class MonetaryAccountReferenceAdapter(converter.JsonAdapter): + @classmethod + def deserialize(cls, target_class, obj): + """ + :type target_class: object_.MonetaryAccountReference|type + :type obj: dict + + :rtype: object_.MonetaryAccountReference + """ + + label_monetary_account = converter.deserialize( + object_.LabelMonetaryAccount, + obj + ) + + return target_class.create_from_label_monetary_account( + label_monetary_account + ) + + @classmethod + def serialize(cls, monetary_account_reference): + """ + :type monetary_account_reference: object_.MonetaryAccountReference + + :rtype: dict + """ + + return converter.serialize(monetary_account_reference.pointer) diff --git a/bunq/sdk/json/pagination_adapter.py b/bunq/sdk/json/pagination_adapter.py new file mode 100644 index 0000000..676e324 --- /dev/null +++ b/bunq/sdk/json/pagination_adapter.py @@ -0,0 +1,108 @@ +import urllib.parse as urlparse + +from bunq import Pagination +from bunq.sdk.json import converter + + +class PaginationAdapter(converter.JsonAdapter): + # Raw pagination response field constants. + _FIELD_FUTURE_URL = 'future_url' + _FIELD_NEWER_URL = 'newer_url' + _FIELD_OLDER_URL = 'older_url' + + # Processed pagination field constants. + _FIELD_OLDER_ID = 'older_id' + _FIELD_NEWER_ID = 'newer_id' + _FIELD_FUTURE_ID = 'future_id' + _FIELD_COUNT = 'count' + + # Very first index in an array. + _INDEX_FIRST = 0 + + @classmethod + def deserialize(cls, target_class, pagination_response): + """ + :type target_class: Pagination|type + :type pagination_response: dict + + :rtype: client.Pagination + """ + + pagination = Pagination() + pagination.__dict__.update( + cls.parse_pagination_dict(pagination_response) + ) + + return pagination + + @classmethod + def parse_pagination_dict(cls, response_obj): + """ + :type response_obj: dict + + :rtype: dict + """ + + pagination_dict = {} + + cls.update_dict_id_field_from_response_field( + pagination_dict, + cls._FIELD_OLDER_ID, + response_obj, + cls._FIELD_OLDER_URL, + Pagination.PARAM_OLDER_ID + ) + cls.update_dict_id_field_from_response_field( + pagination_dict, + cls._FIELD_NEWER_ID, + response_obj, + cls._FIELD_NEWER_URL, + Pagination.PARAM_NEWER_ID + ) + cls.update_dict_id_field_from_response_field( + pagination_dict, + cls._FIELD_FUTURE_ID, + response_obj, + cls._FIELD_FUTURE_URL, + Pagination.PARAM_NEWER_ID + ) + + return pagination_dict + + @classmethod + def update_dict_id_field_from_response_field(cls, dict_, dict_id_field, + response_obj, response_field, + response_param): + """ + :type dict_: dict + :type dict_id_field: str + :type response_obj: dict + :type response_field: str + :type response_param: str + """ + + url = response_obj[response_field] + + if url is not None: + url_parsed = urlparse.urlparse(url) + parameters = urlparse.parse_qs(url_parsed.query) + dict_[dict_id_field] = int( + parameters[response_param][cls._INDEX_FIRST] + ) + + if cls._FIELD_COUNT in parameters and cls._FIELD_COUNT not in dict_: + dict_[cls._FIELD_COUNT] = int( + parameters[Pagination.PARAM_COUNT][cls._INDEX_FIRST] + ) + + @classmethod + def serialize(cls, pagination): + """ + :type pagination: Pagination + + :raise: NotImplementedError + """ + + _ = pagination + + raise NotImplementedError() diff --git a/bunq/sdk/json/session_server_adapter.py b/bunq/sdk/json/session_server_adapter.py new file mode 100644 index 0000000..588e66d --- /dev/null +++ b/bunq/sdk/json/session_server_adapter.py @@ -0,0 +1,108 @@ +from bunq.sdk.exception.bunq_exception import BunqException +from bunq.sdk.json import converter +from bunq.sdk.model.core.id import Id +from bunq.sdk.model.core.session_token import SessionToken +from bunq.sdk.model.generated import endpoint + + +class SessionServerAdapter(converter.JsonAdapter): + # Error constants. + _ERROR_COULD_NOT_DETERMINE_USER = 'Could not determine user.' + + # Id constants + _ATTRIBUTE_ID = '_id_' + _INDEX_ID = 0 + _FIELD_ID = 'Id' + + # Token constants + _ATTRIBUTE_TOKEN = '_token' + _INDEX_TOKEN = 1 + _FIELD_TOKEN = 'Token' + + # User constants + _INDEX_USER = 2 + + # UserCompany constants + _ATTRIBUTE_USER_COMPANY = '_user_company' + _FIELD_USER_COMPANY = 'UserCompany' + + # UserPerson constants + _ATTRIBUTE_USER_PERSON = '_user_person' + _FIELD_USER_PERSON = 'UserPerson' + + # UserApiKey constants + _ATTRIBUTE_USER_API_KEY = '_user_api_key' + _FIELD_USER_API_KEY = 'UserApiKey' + + @classmethod + def deserialize(cls, target_class, array): + """ + :type target_class: SessionServer|type + :type array: list + + :rtype: SessionServer + """ + + session_server = target_class.__new__(target_class) + session_server.__dict__ = { + cls._ATTRIBUTE_ID: converter.deserialize( + Id, + array[cls._INDEX_ID][cls._FIELD_ID] + ), + cls._ATTRIBUTE_TOKEN: converter.deserialize( + SessionToken, + array[cls._INDEX_TOKEN][cls._FIELD_TOKEN] + ), + cls._ATTRIBUTE_USER_COMPANY: None, + cls._ATTRIBUTE_USER_PERSON: None, + } + + user_dict_wrapped = array[cls._INDEX_USER] + + if cls._FIELD_USER_COMPANY in user_dict_wrapped: + session_server.__dict__[cls._ATTRIBUTE_USER_COMPANY] = \ + converter.deserialize( + endpoint.UserCompany, + user_dict_wrapped[cls._FIELD_USER_COMPANY] + ) + elif cls._FIELD_USER_PERSON in user_dict_wrapped: + session_server.__dict__[cls._ATTRIBUTE_USER_PERSON] = \ + converter.deserialize( + endpoint.UserPerson, + user_dict_wrapped[cls._FIELD_USER_PERSON] + ) + elif cls._FIELD_USER_API_KEY in user_dict_wrapped: + session_server.__dict__[cls._ATTRIBUTE_USER_API_KEY] = \ + converter.deserialize( + endpoint.UserApiKey, + user_dict_wrapped[cls._FIELD_USER_API_KEY] + ) + else: + raise BunqException(cls._ERROR_COULD_NOT_DETERMINE_USER) + + return session_server + + @classmethod + def serialize(cls, session_server): + """ + :type session_server: SessionServer + + :rtype: list + """ + + return [ + {cls._FIELD_ID: converter.serialize(session_server.id_)}, + {cls._FIELD_TOKEN: converter.serialize(session_server.token)}, + { + cls._FIELD_USER_COMPANY: + converter.serialize(session_server.user_company), + }, + { + cls._FIELD_USER_PERSON: + converter.serialize(session_server.user_person), + }, + { + cls._FIELD_USER_API_KEY: + converter.serialize(session_server.user_api_key), + }, + ] diff --git a/bunq/sdk/json/share_detail_adapter.py b/bunq/sdk/json/share_detail_adapter.py new file mode 100644 index 0000000..281919e --- /dev/null +++ b/bunq/sdk/json/share_detail_adapter.py @@ -0,0 +1,71 @@ +from bunq.sdk.json import converter +from bunq.sdk.model.generated import object_ + + +class ShareDetailAdapter(converter.JsonAdapter): + # Attribute/Field constants + _ATTRIBUTE_PAYMENT = 'payment' + _FIELD_PAYMENT = 'ShareDetailPayment' + + _ATTRIBUTE_READ_ONLY = 'read_only' + _FIELD_READ_ONLY = 'ShareDetailReadOnly' + + _ATTRIBUTE_DRAFT_PAYMENT = 'draft_payment' + _FIELD_DRAFT_PAYMENT = 'ShareDetailDraftPayment' + + @classmethod + def deserialize(cls, target_class, obj): + """ + :type target_class: object_.ShareDetail|type + :type obj: dict + + :rtype: object_.ShareDetail + """ + + share_detail = target_class.__new__(target_class) + share_detail.__dict__ = { + cls._ATTRIBUTE_PAYMENT: converter.deserialize( + object_.ShareDetailPayment, + cls._get_field_or_none(cls._FIELD_DRAFT_PAYMENT, obj) + ), + cls._ATTRIBUTE_READ_ONLY: converter.deserialize( + object_.ShareDetailReadOnly, + cls._get_field_or_none(cls._FIELD_READ_ONLY, obj) + ), + cls._ATTRIBUTE_DRAFT_PAYMENT: converter.deserialize( + object_.ShareDetailDraftPayment, + cls._get_field_or_none(cls._FIELD_DRAFT_PAYMENT, obj) + ), + } + + return share_detail + + @staticmethod + def _get_field_or_none(field, obj): + """ + :type field: str + :type obj: dict + + :return: dict|None + """ + + return obj[field] if field in obj else None + + @classmethod + def serialize(cls, share_detail): + """ + :type share_detail: object_.ShareDetail + + :rtype: dict + """ + + return { + cls._FIELD_PAYMENT: converter.serialize( + share_detail._payment_field_for_request), + cls._FIELD_READ_ONLY: converter.serialize( + share_detail._read_only_field_for_request), + cls._FIELD_DRAFT_PAYMENT: converter.serialize( + share_detail._draft_payment + ), + } +