service/src/piracyshield_service/ticket/item/ipv6/get_all_by_ticket_checksum.py

70 lines
1.8 KiB
Python

from __future__ import annotations
from piracyshield_service.base import BaseService
from piracyshield_component.security.checksum import Checksum
from piracyshield_component.exception import ApplicationException
from piracyshield_data_storage.ticket.item.storage import TicketItemStorage, TicketItemStorageGetException
from piracyshield_service.ticket.item.errors import TicketItemErrorCode, TicketItemErrorMessage
class TicketItemIPv6GetAllByTicketChecksumService(BaseService):
"""
Returns the checksum of the IPv6 list of a ticket.
"""
checksum = None
data_storage = None
def __init__(self):
"""
Inizialize logger and required modules.
"""
super().__init__()
self._prepare_modules()
def execute(self, ticket_id: str) -> list | Exception:
"""
Get the checksum of all the IPv6 items by ticket_id.
:return
"""
response = self.get_ipv6_all_by_ticket(ticket_id)
# we don't have any data to work on
if not len(response):
raise ApplicationException(TicketItemErrorCode.TICKET_ITEM_EMPTY_CHECKSUM, TicketItemErrorMessage.TICKET_ITEM_EMPTY_CHECKSUM)
data = '\n'.join(response)
try:
return self.checksum.from_string(
algorithm = 'sha256',
string = data
)
except ChecksumUnicodeException as e:
self.logger.error(f'Could not generate the checksum for IPv6 for ticket `{ticket_id}`')
raise ApplicationException(TicketItemErrorCode.GENERIC, TicketItemErrorMessage.GENERIC, e)
def _schedule_task(self):
pass
def _validate_parameters(self):
pass
def _prepare_configs(self):
pass
def _prepare_modules(self):
self.data_storage = TicketItemStorage()
self.checksum = Checksum()