dschick/git_brcal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added validation rules regarding port_id (for berth and participant assignment)

Browse Source
This commit is contained in:
Daniel Schick 2024-10-23 09:55:56 +02:00
parent 0d7861ec36
commit 72d3ad05bf
4 changed files with 43 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/server/BreCal/database/sql_utils.py
View File

@ -27,4 +27,10 @@ def get_shipcall_data_for_id(shipcall_id:int):
"""helper function to load previous shipcall data from the database"""
query = "SELECT * FROM shipcall where id = ?id?"
pdata = execute_sql_query_standalone(query=query, param={"id":shipcall_id}, command_type="single_or_none")
return pdata
def get_port_ids_for_participant_id(participant_id:int):
"""helper function to load all port ids for a participant"""
query = "SELECT port_id FROM participant_port_map where participant_id = ?participant_id?"
pdata = execute_sql_query_standalone(query=query, param={"participant_id":participant_id})
return pdata

11
src/server/BreCal/validators/input_validation_shipcall.py
View File

@ -237,21 +237,22 @@ class InputValidationShipcall():
ship_id = loadedModel.get("ship_id", None)
arrival_berth_id = loadedModel.get("arrival_berth_id", None)
departure_berth_id = loadedModel.get("departure_berth_id", None)
port_id = loadedModel.get("port_id", None)
participants = loadedModel.get("participants",[])
valid_ship_id = check_if_ship_id_is_valid(ship_id=ship_id)
if not valid_ship_id:
raise ValidationError({"ship_id":f"provided an invalid ship id, which is not found in the database: {ship_id}"})
valid_arrival_berth_id = check_if_berth_id_is_valid(berth_id=arrival_berth_id)
valid_arrival_berth_id = check_if_berth_id_is_valid(berth_id=arrival_berth_id, port_id=port_id)
if not valid_arrival_berth_id:
raise ValidationError({"arrival_berth_id":f"provided an invalid arrival berth id, which is not found in the database: {arrival_berth_id}"})
raise ValidationError({"arrival_berth_id":f"provided an invalid arrival berth id, which is not found in the database: {arrival_berth_id}, or the berth is not assigned to the port: {port_id}"})
valid_departure_berth_id = check_if_berth_id_is_valid(berth_id=departure_berth_id)
valid_departure_berth_id = check_if_berth_id_is_valid(berth_id=departure_berth_id, port_id=port_id)
if not valid_departure_berth_id:
raise ValidationError({"departure_berth_id":f"provided an invalid departure berth id, which is not found in the database: {departure_berth_id}"})
raise ValidationError({"departure_berth_id":f"provided an invalid departure berth id, which is not found in the database: {departure_berth_id}, or the berth is not assigned to the port: {port_id}"})
valid_participant_ids = check_if_participant_ids_are_valid(participants=participants)
valid_participant_ids = check_if_participant_ids_are_valid(participants=participants, port_id=port_id)
if not valid_participant_ids:
raise ValidationError({"participants":f"one of the provided participant ids is invalid. Could not find one of these in the database: {participants}"})

5
src/server/BreCal/validators/input_validation_times.py
View File

@ -199,10 +199,11 @@ class InputValidationTimes():
"""
# extract the IDs
berth_id, shipcall_id, participant_id = content.get("berth_id"), content.get("shipcall_id"), content.get("participant_id")
port_id = content.get("port_id", None)
valid_berth_id_reference = check_if_berth_id_is_valid(berth_id)
valid_berth_id_reference = check_if_berth_id_is_valid(berth_id, port_id)
if not valid_berth_id_reference:
raise ValidationError({"berth_id":f"The referenced berth_id '{berth_id}' does not exist in the database."})
raise ValidationError({"berth_id":f"The referenced berth_id '{berth_id}' does not exist in the database or is not assigned to the port '{port_id}'."})
valid_shipcall_id_reference = check_if_shipcall_id_is_valid(shipcall_id)
if not valid_shipcall_id_reference:

44
src/server/BreCal/validators/input_validation_utils.py
View File

@ -8,12 +8,13 @@ from BreCal.impl.berths import GetBerths
from BreCal.impl.shipcalls import GetShipcalls
from BreCal.database.enums import ParticipantType
from BreCal.database.sql_utils import get_port_ids_for_participant_id
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).
of common participant data (not a data model).
"""
# get all participants
response,status_code,header = GetParticipant(options={})
@ -72,7 +73,7 @@ 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.
Note: ParticipantType is an IntFlag.
Hence, ParticipantType(1) is ParticipantType.BSMD,
and ParticipantType(7) is [ParticipantType.BSMD, ParticipantType.TERMINAL, ParticipantType.PILOT]
@ -84,7 +85,7 @@ def check_if_user_is_bsmd_type(user_data:dict)->bool:
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
is_bsmd = ParticipantType.BSMD in participant_type
return is_bsmd
def check_if_user_has_bsmd_flag(user_data:dict)->bool:
@ -92,7 +93,7 @@ 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.
Note: ParticipantType is an IntFlag.
Hence, ParticipantType(1) is ParticipantType.BSMD,
and ParticipantType(7) is [ParticipantType.BSMD, ParticipantType.TERMINAL, ParticipantType.PILOT]
@ -104,7 +105,7 @@ def check_if_user_has_bsmd_flag(user_data:dict)->bool:
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
is_bsmd = ParticipantType.BSMD in participant_type
return is_bsmd
@ -112,31 +113,37 @@ 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):
def check_if_berth_id_is_valid(berth_id, port_id=None):
"""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())
if port_id is not None:
# check, whether the berth is assigned to the respective port
berth_is_assigned_to_port = berths.get(berth_id,{}).get("port_id") == port_id
berth_id_is_valid = berth_id_is_valid and berth_is_assigned_to_port
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()
@ -149,7 +156,7 @@ def check_if_participant_id_is_valid_standalone(participant_id:int, participant_
"""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()
@ -173,10 +180,10 @@ def check_if_participant_id_is_valid_standalone(participant_id:int, participant_
# 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):
def check_if_participant_id_is_valid(participant:dict, port_id=None):
"""
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
@ -185,9 +192,14 @@ def check_if_participant_id_is_valid(participant:dict):
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)
if participant_id_is_valid and port_id is not None:
# check, whether the participant is assigned to the respective port
participant_is_assigned_to_port = any(p.get('port_id', None) == port_id for p in get_port_ids_for_participant_id(participant_id))
participant_id_is_valid &= participant_is_assigned_to_port
return participant_id_is_valid
def check_if_participant_ids_are_valid(participants:list[dict]):
def check_if_participant_ids_are_valid(participants:list[dict], port_id=None):
"""
args:
@ -199,9 +211,9 @@ def check_if_participant_ids_are_valid(participants:list[dict]):
# 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]
valid_participant_ids = [check_if_participant_id_is_valid(participant, port_id) for participant in participants]
# boolean check, whether all participant ids are valid
return all(valid_participant_ids)