import logging import json from collections import Counter from BreCal.impl.participant import GetParticipant from BreCal.impl.ships import GetShips from BreCal.impl.berths import GetBerths from BreCal.impl.shipcalls import GetShipcalls from BreCal.database.enums import ParticipantType from marshmallow import ValidationError def get_participant_id_dictionary(): """ get a dictionary of all participants, where the key is the participant's id, and the value is a dictionary of common participant data (not a data model). """ # get all participants response,status_code,header = GetParticipant(options={}) # build a dictionary of id:item pairs, so one can select the respective participant participants = json.loads(response) participants = {items.get("id"):{**items, "participant_id":items.get("id")} for items in participants} assert all(["participant_id" in participant for k,participant in participants.items()]) return participants def get_berth_id_dictionary(): # get all berths response,status_code,header = GetBerths(token=None) # build a dictionary of id:item pairs, so one can select the respective participant berths = json.loads(response) berths = {items.get("id"):items for items in berths} return berths def get_ship_id_dictionary(): # get all ships response,status_code,header = GetShips(token=None) # build a dictionary of id:item pairs, so one can select the respective participant ships = json.loads(response) ships = {items.get("id"):items for items in ships} return ships def get_shipcall_id_dictionary(): # get all ships response,status_code,header = GetShipcalls(options={'past_days':30000}) # build a dictionary of id:item pairs, so one can select the respective participant shipcalls = json.loads(response) shipcalls = {items.get("id"):items for items in shipcalls} return shipcalls def get_participant_type_from_participant_id(participant_id:int)->ParticipantType: # build a dictionary of id:item pairs, so one can select the respective participant participants = get_participant_id_dictionary() participant = participants.get(participant_id,{}) participant_type = ParticipantType(participant.get("type",0)) return participant_type def get_participant_type_from_user_data(user_data:dict)->ParticipantType: # user_data = decode token participant_id = user_data.get("participant_id") # builds an internal dictionary of id:item pairs, so one can select the respective participant participant_type = get_participant_type_from_participant_id(participant_id) return participant_type def check_if_user_is_bsmd_type(user_data:dict)->bool: """ given a dictionary of user data, determine the respective participant id and read, whether that participant is a .BSMD-type Note: ParticipantType is an IntFlag. Hence, ParticipantType(1) is ParticipantType.BSMD, and ParticipantType(7) is [ParticipantType.BSMD, ParticipantType.TERMINAL, ParticipantType.PILOT] both would return 'True' returns: boolean. Whether the participant id is a .BSMD type element """ # use the decoded JWT token and extract the participant type participant_type = get_participant_type_from_user_data(user_data) # boolean check: is the participant of type .BSMD? is_bsmd = ParticipantType.BSMD in participant_type return is_bsmd def check_if_user_has_bsmd_flag(user_data:dict)->bool: """ given a dictionary of user data, determine the respective participant id and read, whether that participant is a .BSMD-type Note: ParticipantType is an IntFlag. Hence, ParticipantType(1) is ParticipantType.BSMD, and ParticipantType(7) is [ParticipantType.BSMD, ParticipantType.TERMINAL, ParticipantType.PILOT] both would return 'True' returns: boolean. Whether the participant id is a .BSMD type element """ # use the decoded JWT token and extract the participant type participant_type = get_participant_type_from_user_data(user_data) # boolean check: is the participant of type .BSMD? is_bsmd = ParticipantType.BSMD in participant_type return is_bsmd def check_if_ship_id_is_valid(ship_id): """check, whether the provided ID is valid. If it is 'None', it will be considered valid. This is, because a shipcall POST-request, does not have to include all IDs at once""" if ship_id is None: return True # build a dictionary of id:item pairs, so one can select the respective participant ships = get_ship_id_dictionary() # boolean check ship_id_is_valid = ship_id in list(ships.keys()) return ship_id_is_valid def check_if_berth_id_is_valid(berth_id): """check, whether the provided ID is valid. If it is 'None', it will be considered valid. This is, because a shipcall POST-request, does not have to include all IDs at once""" if berth_id is None: return True # build a dictionary of id:item pairs, so one can select the respective participant berths = get_berth_id_dictionary() # boolean check berth_id_is_valid = berth_id in list(berths.keys()) return berth_id_is_valid def check_if_shipcall_id_is_valid(shipcall_id:int): """check, whether the provided ID is valid. If it is 'None', it will be considered valid. This is, because a request, may not have to include all IDs at once""" if shipcall_id is None: return True # build a dictionary of id:item pairs, so one can select the respective participant shipcalls = get_shipcall_id_dictionary() # boolean check shipcall_id_is_valid = shipcall_id in list(shipcalls.keys()) return shipcall_id_is_valid import typing def check_if_participant_id_is_valid_standalone(participant_id:int, participant_type:typing.Optional[ParticipantType]): """check, whether the provided ID is valid. If it is 'None', it will be considered valid. This is, because a request, may not have to include all IDs at once""" if participant_id is None: return True # build a dictionary of id:item pairs, so one can select the respective participant participants = get_participant_id_dictionary() # boolean check participant_id_is_valid = participant_id in list(participants.keys()) if participant_type is not None: if participant_id not in list(participants.keys()): raise ValidationError(f"the provided participant_id {participant_id} does not exist in the database.") # IntFlag object participant_type_in_db = ParticipantType(int(participants.get(participant_id).get("type", ParticipantType.undefined))) assert isinstance(participant_type_in_db, ParticipantType), f"{type(participant_type_in_db)}" # IntFlag comparison. A user may be assigned as a pilot, but the participant may be multiple roles participant_type_matches_db = (participant_type in participant_type_in_db) participant_is_valid = (participant_id_is_valid and participant_type_matches_db) return participant_is_valid else: # when the participant_type is not provided, only evaluate the ID return participant_id_is_valid def check_if_participant_id_is_valid(participant:dict): """ check, whether the provided ID is valid. If it is 'None', it will be considered valid. This is, because a shipcall POST-request, does not have to include all IDs at once Following the common BreCal.schemas.model.ParticipantAssignmentSchema, a participant dictionary contains the keys: 'participant_id' : int 'type' : ParticipantType """ # #TODO1: Daniel Schick: 'types may only appear once and must not include type "BSMD"' participant_id = participant.get("participant_id", None) participant_type = ParticipantType(int(participant.get("type", ParticipantType.undefined))) participant_id_is_valid = check_if_participant_id_is_valid_standalone(participant_id, participant_type) return participant_id_is_valid def check_if_participant_ids_are_valid(participants:list[dict]): """ args: participants (list of participant-elements) Following the common BreCal.schemas.model.ParticipantAssignmentSchema, a participant dictionary contains the keys: 'participant_id' : int 'type' : ParticipantType """ # empty list -> invalid if participants is None: return False # check each participant id individually valid_participant_ids = [check_if_participant_id_is_valid(participant) for participant in participants] # boolean check, whether all participant ids are valid return all(valid_participant_ids) def check_if_participant_ids_and_types_are_valid(participants:list[dict[str,int]]): # creates a Counter object, which counts the number of unique elements # key of counter: type, value of counter: number of listings in 'participants' # e.g., {1: 4, 2: 1, 8: 1} (type 1 occurs 4 times in this example) counter_type = Counter([participant.get("type") for participant in participants]) counter_id = Counter([participant.get("type") for participant in participants]) # obtains the maximum count from the counter's values max_count_type = max(list(counter_type.values())) if len(list(counter_type.values()))>0 else 0 max_count_ids = max(list(counter_id.values())) if len(list(counter_id.values()))>0 else 0 # when 0 or 1 count for the participant ids or types, return true. Return false, when there is more than one entry. return max_count_type <= 1 and max_count_ids <= 1