Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _from_data_v1(cls, data):
if data["version"] == 1:
segs = SegmentSequence(data['segments_bin']).segments
return cls(segs[0], segs[1], data['resume_method'], data['tan_request_structured'])
raise Exception("Wrong blob data version")
def get_data(self) -> bytes:
"""Return a compressed datablob representing this object.
To restore the object, use :func:`fints.client.NeedRetryResponse.from_data`.
"""
data = {
"_class_name": self.__class__.__name__,
"version": 1,
"segments_bin": SegmentSequence([self.command_seg, self.tan_request]).render_bytes(),
"resume_method": self.resume_method,
"tan_request_structured": self.tan_request_structured,
}
return compress_datablob(DATA_BLOB_MAGIC_RETRY, 1, data)
self.bank_identifier = BankIdentifier(BankIdentifier.COUNTRY_ALPHA_TO_NUMERIC['DE'], bank_identifier)
else:
raise TypeError("bank_identifier must be BankIdentifier or str (BLZ)")
self.system_id = SYSTEM_ID_UNASSIGNED
if not product_id:
logger.warning('You should register your program with the ZKA and pass your own product_id as a parameter.')
product_id = 'DC333D745719C4BD6A6F9DB6A'
self.user_id = user_id
self.customer_id = customer_id or user_id
self.bpd_version = 0
self.bpa = None
self.bpd = SegmentSequence()
self.upd_version = 0
self.upa = None
self.upd = SegmentSequence()
self.product_name = product_id
self.product_version = product_version
self.response_callbacks = []
self.mode = mode
self.init_tan_response = None
self._standing_dialog = None
if from_data:
self.set_data(bytes(from_data))
def _set_data_v1(self, data):
self.system_id = data.get('system_id', self.system_id)
if all(x in data for x in ('bpd_bin', 'bpa_bin', 'bpd_version')):
if data['bpd_version'] >= self.bpd_version and data['bpa_bin']:
self.bpd = SegmentSequence(data['bpd_bin'])
self.bpa = SegmentSequence(data['bpa_bin']).segments[0]
self.bpd_version = data['bpd_version']
if all(x in data for x in ('upd_bin', 'upa_bin', 'upd_version')):
if data['upd_version'] >= self.upd_version and data['upa_bin']:
self.upd = SegmentSequence(data['upd_bin'])
self.upa = SegmentSequence(data['upa_bin']).segments[0]
self.upd_version = data['upd_version']
def process_response_message(self, dialog, message: FinTSInstituteMessage, internal_send=True):
bpa = message.find_segment_first(HIBPA3)
if bpa:
self.bpa = bpa
self.bpd_version = bpa.bpd_version
self.bpd = SegmentSequence(
message.find_segments(
callback=lambda m: len(m.header.type) == 6 and m.header.type[1] == 'I' and m.header.type[5] == 'S'
)
)
upa = message.find_segment_first(HIUPA4)
if upa:
self.upa = upa
self.upd_version = upa.upd_version
self.upd = SegmentSequence(
message.find_segments('HIUPD')
)
for seg in message.find_segments(HIRMG2):
for response in seg.responses:
if not internal_send:
self._log_response(None, response)
self._call_callbacks(None, response)
self._process_response(dialog, None, response)
for seg in message.find_segments(HIRMS2):
for response in seg.responses:
segment = None # FIXME: Provide segment
def _parse_value(self, value):
if isinstance(value, SegmentSequence):
return value
else:
return SegmentSequence(value)
def _set_data_v1(self, data):
self.system_id = data.get('system_id', self.system_id)
if all(x in data for x in ('bpd_bin', 'bpa_bin', 'bpd_version')):
if data['bpd_version'] >= self.bpd_version and data['bpa_bin']:
self.bpd = SegmentSequence(data['bpd_bin'])
self.bpa = SegmentSequence(data['bpa_bin']).segments[0]
self.bpd_version = data['bpd_version']
if all(x in data for x in ('upd_bin', 'upa_bin', 'upd_version')):
if data['upd_version'] >= self.upd_version and data['upa_bin']:
self.upd = SegmentSequence(data['upd_bin'])
self.upa = SegmentSequence(data['upa_bin']).segments[0]
self.upd_version = data['upd_version']
def _set_data_v1(self, data):
self.system_id = data.get('system_id', self.system_id)
if all(x in data for x in ('bpd_bin', 'bpa_bin', 'bpd_version')):
if data['bpd_version'] >= self.bpd_version and data['bpa_bin']:
self.bpd = SegmentSequence(data['bpd_bin'])
self.bpa = SegmentSequence(data['bpa_bin']).segments[0]
self.bpd_version = data['bpd_version']
if all(x in data for x in ('upd_bin', 'upa_bin', 'upd_version')):
if data['upd_version'] >= self.upd_version and data['upa_bin']:
self.upd = SegmentSequence(data['upd_bin'])
self.upa = SegmentSequence(data['upa_bin']).segments[0]
self.upd_version = data['upd_version']
),
key_name=KeyName(
message.dialog.client.bank_identifier,
message.dialog.client.user_id,
KeyType.V,
0,
0,
),
compression_function=CompressionFunction.NULL,
)
)
message.segments[1].header.number = 998
message.segments.insert(
2,
HNVSD1(
data=SegmentSequence(segments=plain_segments)
)
)
message.segments[2].header.number = 999
def process_response_message(self, dialog, message: FinTSInstituteMessage, internal_send=True):
bpa = message.find_segment_first(HIBPA3)
if bpa:
self.bpa = bpa
self.bpd_version = bpa.bpd_version
self.bpd = SegmentSequence(
message.find_segments(
callback=lambda m: len(m.header.type) == 6 and m.header.type[1] == 'I' and m.header.type[5] == 'S'
)
)
upa = message.find_segment_first(HIUPA4)
if upa:
self.upa = upa
self.upd_version = upa.upd_version
self.upd = SegmentSequence(
message.find_segments('HIUPD')
)
for seg in message.find_segments(HIRMG2):
for response in seg.responses:
if not internal_send: